From e8882b980944ee22f4b9d7e281a78a490265d3f6 Mon Sep 17 00:00:00 2001 From: Gabe Date: Tue, 4 Aug 2026 10:36:16 -0400 Subject: [PATCH] route limited mailboxes through uidonly --- crates/admin/src/meta.rs | 2 + crates/core/src/account/migration.rs | 34 + crates/core/src/account/payload.rs | 5 + crates/core/src/account/view.rs | 2 + crates/core/src/cache/imap/download/flow.rs | 247 ++- crates/core/src/cache/imap/download/mod.rs | 11 +- .../core/src/cache/imap/download/rebuild.rs | 91 +- crates/core/src/cache/imap/mailbox.rs | 4 + crates/core/src/imap/client.rs | 17 +- crates/core/src/imap/manager.rs | 55 +- crates/core/src/imap/mock_server.rs | 16 +- crates/core/src/imap/mod.rs | 1 + crates/core/src/imap/uidonly_acquisition.rs | 1823 +++++++++++++++++ crates/core/src/import/mod.rs | 2 + crates/smtp/src/server.rs | 1 + web/src/api/account/api.ts | 3 +- web/src/features/accounts/account-new.tsx | 2 + .../accounts/account-settings-page.tsx | 2 + .../components/__tests__/schema-edge.test.ts | 1 + .../components/__tests__/schema.test.ts | 1 + .../features/accounts/components/schema.ts | 1 + .../accounts/components/tab-server.tsx | 16 + web/src/locales/en.json | 4 +- 23 files changed, 2243 insertions(+), 98 deletions(-) create mode 100644 crates/core/src/imap/uidonly_acquisition.rs diff --git a/crates/admin/src/meta.rs b/crates/admin/src/meta.rs index dfd91181..4f1f7e44 100644 --- a/crates/admin/src/meta.rs +++ b/crates/admin/src/meta.rs @@ -256,6 +256,7 @@ impl From for AccountModel { updated_at: value.updated_at, created_by: value.created_by, use_dangerous: value.use_dangerous, + uidonly_enabled: false, pgp_key: value.pgp_key, imap_quota_window: None, imap_quota_bytes: None, @@ -666,6 +667,7 @@ impl From for bichon_core::cache::imap::mailbox::MailBox { uid_next: value.uid_next, uid_validity: value.uid_validity, highest_uid: None, + uidonly_source_scope: None, } } } diff --git a/crates/core/src/account/migration.rs b/crates/core/src/account/migration.rs index 531f701e..5c31020d 100644 --- a/crates/core/src/account/migration.rs +++ b/crates/core/src/account/migration.rs @@ -301,6 +301,9 @@ pub struct Account { pub updated_at: i64, pub created_by: u64, //user id pub use_dangerous: bool, + /// UIDONLY acquisition is opt-in because it changes remote message identity. + #[serde(default)] + pub uidonly_enabled: bool, pub pgp_key: Option, pub imap_quota_bytes: Option, pub imap_quota_window: Option, @@ -345,6 +348,7 @@ impl Account { created_at: utc_now!(), updated_at: utc_now!(), use_dangerous: request.use_dangerous, + uidonly_enabled: request.uidonly_enabled.unwrap_or_default(), pgp_key: request.pgp_key, created_by: user_id, download_batch_size: request.download_batch_size, @@ -668,6 +672,10 @@ impl Account { new.use_dangerous = use_dangerous; } + if let Some(uidonly_enabled) = request.uidonly_enabled { + new.uidonly_enabled = uidonly_enabled; + } + if let Some(pgp_key) = request.pgp_key { new.pgp_key = Some(pgp_key); } @@ -704,6 +712,32 @@ impl Account { mod tests { use super::*; + #[test] + fn uidonly_setting_defaults_off_and_only_changes_explicitly() { + let account = Account::new(1, AccountCreateRequest::default()).unwrap(); + assert!(!account.uidonly_enabled); + let update = |old: &Account, value| { + Account::apply_update_fields( + old, + AccountUpdateRequest { + uidonly_enabled: value, + ..Default::default() + }, + ) + .unwrap() + }; + let enabled = update(&account, Some(true)); + assert!(enabled.uidonly_enabled); + assert!(update(&enabled, None).uidonly_enabled); + assert!(!update(&enabled, Some(false)).uidonly_enabled); + + let mut legacy = serde_json::to_value(enabled).unwrap(); + legacy.as_object_mut().unwrap().remove("uidonly_enabled"); + assert!(!serde_json::from_value::(legacy) + .unwrap() + .uidonly_enabled); + } + // ── FilterRule ─────────────────────────────────────────────────── #[test] diff --git a/crates/core/src/account/payload.rs b/crates/core/src/account/payload.rs index a0c55943..6b1fd929 100644 --- a/crates/core/src/account/payload.rs +++ b/crates/core/src/account/payload.rs @@ -52,6 +52,9 @@ pub struct AccountCreateRequest { pub download_batch_size: Option, pub max_email_size_bytes: Option, pub use_dangerous: bool, + /// Explicitly enables UIDONLY acquisition when required by the server. + /// Omitted values remain disabled for backward compatibility. + pub uidonly_enabled: Option, pub pgp_key: Option, pub imap_quota_bytes: Option, pub imap_quota_window: Option, @@ -188,6 +191,8 @@ pub struct AccountUpdateRequest { pub download_batch_size: Option, pub max_email_size_bytes: Option, pub use_dangerous: Option, + /// Enables or disables UIDONLY acquisition for this account. + pub uidonly_enabled: Option, pub pgp_key: Option, pub imap_quota_bytes: Option, diff --git a/crates/core/src/account/view.rs b/crates/core/src/account/view.rs index d800f0b0..f585e29c 100644 --- a/crates/core/src/account/view.rs +++ b/crates/core/src/account/view.rs @@ -52,6 +52,7 @@ pub struct AccountResp { pub created_user_name: String, pub created_user_email: String, pub use_dangerous: bool, + pub uidonly_enabled: bool, pub pgp_key: Option, pub imap_quota_bytes: Option, pub imap_quota_window: Option, @@ -90,6 +91,7 @@ impl AccountResp { .map(|u| u.email.clone()) .unwrap_or_else(|| "N/A".to_string()), use_dangerous: account.use_dangerous, + uidonly_enabled: account.uidonly_enabled, pgp_key: account.pgp_key, imap_quota_bytes: account.imap_quota_bytes, imap_quota_window: account.imap_quota_window, diff --git a/crates/core/src/cache/imap/download/flow.rs b/crates/core/src/cache/imap/download/flow.rs index 3fc0b7bb..4faf4e44 100644 --- a/crates/core/src/cache/imap/download/flow.rs +++ b/crates/core/src/cache/imap/download/flow.rs @@ -35,6 +35,10 @@ use crate::{ imap::executor::{ compress_uid_list, generate_uid_sequence_hashset, ImapExecutor, DEFAULT_BATCH_SIZE, }, + imap::uidonly_acquisition::{ + account_requires_uidonly, connect_and_acquire_or_legacy, + mailbox_has_uidonly_proof, AcquisitionRoute, + }, store::tantivy::envelope::ENVELOPE_MANAGER, }, }; @@ -44,7 +48,6 @@ use tracing::{debug, error, info, warn}; const MAX_NETWORK_RETRIES: u32 = 3; - #[derive(Clone, Debug, Eq, PartialEq)] pub enum FetchDirection { Since, @@ -58,6 +61,12 @@ pub async fn fetch_and_save_by_date( direction: FetchDirection, token: CancellationToken, ) -> BichonResult> { + if account_requires_uidonly(account) || mailbox.uidonly_source_scope.is_some() { + return Err(raise_error!( + "Date-scoped acquisition is unsafe on a UIDONLY-limited server".into(), + ErrorCode::Incompatible + )); + } let account_id = account.id; let mut session = match ImapExecutor::create_connection(account_id).await { Ok(session) => session, @@ -253,19 +262,59 @@ 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. +/// Returns a checkpoint only after the complete mailbox run succeeds. pub async fn fetch_and_save_full_mailbox( account: &AccountModel, mailbox: &MailBox, + force_uidonly: bool, token: CancellationToken, -) -> BichonResult> { +) -> BichonResult { let mailbox_id = mailbox.id; let account_id = account.id; - let mut session = match ImapExecutor::create_connection(account_id).await { - Ok(session) => session, + // Classify the actual acquisition connection. Cached capabilities route + // known-limited accounts early, but can never authorize a legacy fetch. + let route = connect_and_acquire_or_legacy( + account, + mailbox, + force_uidonly, + token.clone(), + |progress| { + DownloadState::update_folder_progress( + account_id, + mailbox.name.clone(), + progress.planned, + progress.resolved, + FolderStatus::Downloading, + None, + ) + }, + ) + .await; + let mut session = match route { + Ok(AcquisitionRoute::Acquired { + report, + source_scope, + }) => { + DownloadState::update_folder_progress( + account_id, + mailbox.name.clone(), + report.inventoried, + report.archived, + FolderStatus::Success, + None, + )?; + let mut updated = mailbox.clone(); + updated.highest_uid = report.checkpoint; + updated.uid_validity = Some(report.uid_validity); + updated.uid_next = Some(report.uid_next); + updated.exists = report.exists; + updated.uidonly_source_scope = Some(source_scope); + return Ok(updated); + } + Ok(AcquisitionRoute::Legacy(session)) => session, Err(e) => { - let err_msg = format!("Connection failed for this folder: {:#?}", e); + let err_msg = format!("Full mailbox acquisition failed: {:#?}", e); DownloadState::update_folder_progress( account_id, mailbox.name.clone(), @@ -310,7 +359,6 @@ pub async fn fetch_and_save_full_mailbox( ); let mut current_processed = 0u64; - let mut has_error_or_cancel = false; let mut max_uid: Option = None; for page in 1..=total_batches { @@ -328,8 +376,11 @@ pub async fn fetch_and_save_full_mailbox( FolderStatus::Cancelled, None, )?; - has_error_or_cancel = true; - break; + session.logout().await.ok(); + return Err(raise_error!( + "Full mailbox download cancelled".into(), + ErrorCode::InternalError + )); } let mut retries = 0u32; @@ -408,24 +459,25 @@ pub async fn fetch_and_save_full_mailbox( FolderStatus::Failed, Some(err_msg), )?; - has_error_or_cancel = true; - break; + session.logout().await.ok(); + return Err(e); } }; } - if !has_error_or_cancel { - DownloadState::update_folder_progress( - account_id, - mailbox.name.clone(), - total, - current_processed, - FolderStatus::Success, - None, - )?; - } + DownloadState::update_folder_progress( + account_id, + mailbox.name.clone(), + total, + current_processed, + FolderStatus::Success, + None, + )?; session.logout().await.ok(); - Ok(max_uid) + let mut updated = mailbox.clone(); + updated.highest_uid = max_uid; + updated.uidonly_source_scope = None; + Ok(updated) } /// Generates a synthetic UIDVALIDITY for IMAP servers that don't provide it. @@ -706,14 +758,37 @@ async fn reconcile_uid_validity_change( Ok(max_uid) } +fn mailbox_source_changed(account: &AccountModel, mailbox: &MailBox) -> bool { + mailbox.uidonly_source_scope.is_some() && !mailbox_has_uidonly_proof(account, mailbox) +} + +fn intersecting_source_changed(account: &AccountModel, mailboxes: &[(MailBox, MailBox)]) -> bool { + mailboxes + .iter() + .any(|(local, _)| mailbox_source_changed(account, local)) +} + pub async fn reconcile_mailboxes( account: &AccountModel, remote_mailboxes: &[MailBox], local_mailboxes: &[MailBox], token: CancellationToken, ) -> BichonResult<()> { - let start_time = Instant::now(); let existing_mailboxes = find_intersecting_mailboxes(local_mailboxes, remote_mailboxes); + let source_changed = intersecting_source_changed(account, &existing_mailboxes); + let known_limited = local_mailboxes + .iter() + .any(|mailbox| mailbox_has_uidonly_proof(account, mailbox)) + || account_requires_uidonly(account); + if (known_limited || source_changed) + && (account.date_since.is_some() || account.date_before.is_some()) + { + return Err(raise_error!( + "Date-scoped acquisition is unsafe on a UIDONLY-limited server".into(), + ErrorCode::Incompatible + )); + } + let start_time = Instant::now(); let account_id = account.id; if !existing_mailboxes.is_empty() { let mut mailboxes_to_update = Vec::with_capacity(existing_mailboxes.len()); @@ -724,6 +799,7 @@ pub async fn reconcile_mailboxes( )?; for (local_mailbox, remote_mailbox) in &existing_mailboxes { + let source_changed = mailbox_source_changed(account, local_mailbox); if token.is_cancelled() { DownloadState::update_session_status( account.id, @@ -733,6 +809,30 @@ pub async fn reconcile_mailboxes( break; } + if account.date_since.is_none() + && account.date_before.is_none() + && (known_limited || source_changed) + { + let can_resume = local_mailbox.uid_validity == remote_mailbox.uid_validity + && !source_changed + && mailbox_has_uidonly_proof(account, local_mailbox); + let mut acquisition_mailbox = (*remote_mailbox).clone(); + if can_resume { + acquisition_mailbox.highest_uid = local_mailbox.highest_uid; + acquisition_mailbox.uidonly_source_scope = + local_mailbox.uidonly_source_scope.clone(); + } + let result = fetch_and_save_full_mailbox( + account, + &acquisition_mailbox, + known_limited, + token.clone(), + ) + .await?; + mailboxes_to_update.push(result); + continue; + } + // Handle missing UIDVALIDITY from non-compliant IMAP servers // (e.g., Tencent Enterprise Mail, etc.) let remote_uid_validity = match remote_mailbox.uid_validity { @@ -838,7 +938,11 @@ pub async fn reconcile_mailboxes( )?; break; } - if mailbox.exists > 0 { + if mailbox.exists > 0 + || (account.date_since.is_none() + && account.date_before.is_none() + && (known_limited || account_requires_uidonly(account))) + { let account = account.clone(); let mailbox = mailbox.clone(); @@ -854,41 +958,50 @@ pub async fn reconcile_mailboxes( }; let result = match &account.date_since { - Some(date_since) => { - rebuild_mailbox_cache_by_date( + Some(date_since) => rebuild_mailbox_cache_by_date( + &account, + mailbox.id, + &date_since.since_date()?, + &mailbox, + FetchDirection::Since, + token.clone(), + ) + .await + .map(|highest_uid| { + let mut updated = mailbox.clone(); + updated.highest_uid = highest_uid; + updated + }), + None => match &account.date_before { + Some(r) => rebuild_mailbox_cache_by_date( &account, mailbox.id, - &date_since.since_date()?, + &r.calculate_date()?, &mailbox, - FetchDirection::Since, + FetchDirection::Before, token.clone(), ) .await - } - None => match &account.date_before { - Some(r) => { - rebuild_mailbox_cache_by_date( + .map(|highest_uid| { + let mut updated = mailbox.clone(); + updated.highest_uid = highest_uid; + updated + }), + None => { + rebuild_mailbox_cache( &account, - mailbox.id, - &r.calculate_date()?, &mailbox, - FetchDirection::Before, + &mailbox, + known_limited, token.clone(), ) .await } - None => { - rebuild_mailbox_cache(&account, &mailbox, &mailbox, token.clone()).await - } }, }; match result { - Ok(new_highest_uid) => { - let mut updated = mailbox.clone(); - updated.highest_uid = new_highest_uid; - MailBox::batch_upsert(&[updated])?; - } + Ok(updated) => MailBox::batch_upsert(&[updated])?, Err(err) => { has_error = true; tracing::error!("Folder sync task failed: {:#?}", err); @@ -975,9 +1088,13 @@ async fn perform_incremental_sync( } None => { fetch_and_save_full_mailbox( - account, remote_mailbox, token, + account, + remote_mailbox, + false, + token, ) .await? + .highest_uid } }, }; @@ -1023,6 +1140,48 @@ mod tests { // Pure unit tests (no network) // ============================================================ + #[test] + fn stale_nonremote_uidonly_scope_does_not_force_current_mailbox_rescan() { + let local = [ + MailBox { + name: "INBOX".into(), + ..Default::default() + }, + MailBox { + name: "Unsubscribed".into(), + uidonly_source_scope: Some("old-source".into()), + ..Default::default() + }, + ]; + let remote = [MailBox { + name: "INBOX".into(), + ..Default::default() + }]; + let current = find_intersecting_mailboxes(&local, &remote); + assert!(mailbox_source_changed(&AccountModel::default(), &local[1])); + assert!(!intersecting_source_changed( + &AccountModel::default(), + ¤t + )); + } + + #[tokio::test] + async fn known_limited_server_rejects_date_scoped_fallback_before_connecting() { + let account = AccountModel { + capabilities: Some(vec!["UIDONLY".into(), "MESSAGELIMIT=10000".into()]), + ..Default::default() + }; + assert!(fetch_and_save_by_date( + &account, + "2026-01-01", + &MailBox::default(), + FetchDirection::Since, + CancellationToken::new(), + ) + .await + .is_err()); + } + #[test] fn test_generate_synthetic_uidvalidity_deterministic() { let a = generate_synthetic_uidvalidity("INBOX"); diff --git a/crates/core/src/cache/imap/download/mod.rs b/crates/core/src/cache/imap/download/mod.rs index c4653f96..03942218 100644 --- a/crates/core/src/cache/imap/download/mod.rs +++ b/crates/core/src/cache/imap/download/mod.rs @@ -78,11 +78,14 @@ pub async fn process_imap_download( } }; session.logout().await.ok(); + // The discovery connection refreshes cached capabilities. Route the + // download from that fresh snapshot, not the stale task input. + let account = AccountModel::get(account_id)?; if matches!(download_task, DownloadTask::FullFetch) { let result = match &account.date_since { Some(date_since) => { rebuild_cache_by_date( - account, + &account, &remote_mailboxes, &date_since.since_date()?, FetchDirection::Since, @@ -93,7 +96,7 @@ pub async fn process_imap_download( None => match &account.date_before { Some(r) => { rebuild_cache_by_date( - account, + &account, &remote_mailboxes, &r.calculate_date()?, FetchDirection::Before, @@ -101,7 +104,7 @@ pub async fn process_imap_download( ) .await } - None => rebuild_cache(account, &remote_mailboxes, token).await, + None => rebuild_cache(&account, &remote_mailboxes, token).await, }, }; match result { @@ -122,7 +125,7 @@ pub async fn process_imap_download( } let local_mailboxes = MailBox::list_all(account_id)?; - match reconcile_mailboxes(account, &remote_mailboxes, &local_mailboxes, token).await { + match reconcile_mailboxes(&account, &remote_mailboxes, &local_mailboxes, token).await { Ok(_) => DownloadState::update_session_status(account_id, DownloadStatus::Success, None)?, Err(e) => { let err_msg = format!("Email Download interrupted: {:#?}", e); diff --git a/crates/core/src/cache/imap/download/rebuild.rs b/crates/core/src/cache/imap/download/rebuild.rs index 96d31d9e..50d19cab 100644 --- a/crates/core/src/cache/imap/download/rebuild.rs +++ b/crates/core/src/cache/imap/download/rebuild.rs @@ -23,19 +23,28 @@ use crate::{ }, cache::{ imap::{ - download::flow::{fetch_and_save_by_date, fetch_and_save_full_mailbox, FetchDirection}, + download::flow::{ + fetch_and_save_by_date, fetch_and_save_full_mailbox, FetchDirection, + }, mailbox::MailBox, }, SEMAPHORE, }, error::{code::ErrorCode, BichonResult}, + imap::uidonly_acquisition::account_requires_uidonly, raise_error, - store::tantivy::{attachment::ATTACHMENT_MANAGER, envelope::ENVELOPE_MANAGER}, + store::tantivy::envelope::ENVELOPE_MANAGER, }; use tokio_util::sync::CancellationToken; use tracing::{error, info}; +fn can_trust_empty_mailbox(account: &AccountModel, mailbox: &MailBox) -> bool { + mailbox.exists == 0 + && mailbox.uidonly_source_scope.is_none() + && !account_requires_uidonly(account) +} + pub async fn rebuild_cache( account: &AccountModel, remote_mailboxes: &[MailBox], @@ -59,7 +68,7 @@ pub async fn rebuild_cache( )?; break; } - if mailbox.exists == 0 { + if can_trust_empty_mailbox(account, mailbox) { info!( "Account {}: Mailbox '{}' on the remote server has no emails. Skipping fetch for this mailbox.", account.id, &mailbox.name @@ -88,12 +97,8 @@ pub async fn rebuild_cache( } }; - match fetch_and_save_full_mailbox(&account, &mailbox, token.clone()).await { - Ok(new_highest_uid) => { - let mut updated = mailbox.clone(); - updated.highest_uid = new_highest_uid; - MailBox::batch_upsert(&[updated])?; - } + match fetch_and_save_full_mailbox(&account, &mailbox, false, token.clone()).await { + Ok(updated) => MailBox::batch_upsert(&[updated])?, Err(err) => { has_error = true; tracing::error!("Folder sync task failed: {:#?}", err); @@ -114,6 +119,29 @@ pub async fn rebuild_cache( Ok(()) } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_ordinary_empty_mailboxes_use_the_fast_path() { + let mailbox = MailBox::default(); + assert!(can_trust_empty_mailbox(&AccountModel::default(), &mailbox)); + let limited = AccountModel { + capabilities: Some(vec!["UIDONLY".into(), "MESSAGELIMIT=10000".into()]), + ..Default::default() + }; + assert!(!can_trust_empty_mailbox(&limited, &mailbox)); + assert!(!can_trust_empty_mailbox( + &AccountModel::default(), + &MailBox { + uidonly_source_scope: Some("proof".into()), + ..Default::default() + } + )); + } +} + pub async fn rebuild_cache_by_date( account: &AccountModel, remote_mailboxes: &[MailBox], @@ -121,6 +149,16 @@ pub async fn rebuild_cache_by_date( direction: FetchDirection, token: CancellationToken, ) -> BichonResult<()> { + if account_requires_uidonly(account) + || remote_mailboxes + .iter() + .any(|mailbox| mailbox.uidonly_source_scope.is_some()) + { + return Err(raise_error!( + "Date-scoped acquisition is unsafe on a UIDONLY-limited server".into(), + ErrorCode::Incompatible + )); + } MailBox::batch_insert(remote_mailboxes)?; DownloadState::init_folder_details( account.id, @@ -203,33 +241,13 @@ pub async fn rebuild_mailbox_cache( account: &AccountModel, local_mailbox: &MailBox, remote_mailbox: &MailBox, + force_uidonly: bool, token: CancellationToken, -) -> BichonResult> { +) -> BichonResult { ENVELOPE_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox.id]) .await?; - ATTACHMENT_MANAGER - .delete_mailbox_attachments(account.id, vec![local_mailbox.id]) - .await?; - if remote_mailbox.exists == 0 { - info!( - "Account {}: Mailbox '{}' has no emails on the remote server. The mailbox is empty, no envelopes to fetch.", - account.id, - &local_mailbox.name - ); - DownloadState::update_folder_progress( - account.id, - remote_mailbox.name.clone(), - 0, - 0, - FolderStatus::Success, - None, - )?; - return Ok(None); - } - - let result = fetch_and_save_full_mailbox(account, remote_mailbox, token).await?; - Ok(result) + fetch_and_save_full_mailbox(account, remote_mailbox, force_uidonly, token).await } pub async fn rebuild_mailbox_cache_by_date( @@ -240,12 +258,15 @@ pub async fn rebuild_mailbox_cache_by_date( direction: FetchDirection, token: CancellationToken, ) -> BichonResult> { + if account_requires_uidonly(account) || remote.uidonly_source_scope.is_some() { + return Err(raise_error!( + "Date-scoped acquisition is unsafe on a UIDONLY-limited server".into(), + ErrorCode::Incompatible + )); + } ENVELOPE_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox_id]) .await?; - ATTACHMENT_MANAGER - .delete_mailbox_attachments(account.id, vec![local_mailbox_id]) - .await?; if remote.exists == 0 { info!( "Account {}: Mailbox '{}' has no emails on the remote server. The mailbox is empty, no envelopes to fetch.", diff --git a/crates/core/src/cache/imap/mailbox.rs b/crates/core/src/cache/imap/mailbox.rs index db11e4e2..69485c18 100644 --- a/crates/core/src/cache/imap/mailbox.rs +++ b/crates/core/src/cache/imap/mailbox.rs @@ -60,6 +60,10 @@ pub struct MailBox { /// Used for incremental sync: next fetch starts from `highest_uid + 1`. /// If `None`, a fallback query against the Tantivy index will be performed once. pub highest_uid: Option, + /// Source fingerprint proving that `highest_uid` came from a complete + /// UIDONLY traversal rather than a provider-limited legacy view. + #[serde(default)] + pub uidonly_source_scope: Option, } impl MemDbModel for MailBox { diff --git a/crates/core/src/imap/client.rs b/crates/core/src/imap/client.rs index 85cbceea..00ec2311 100644 --- a/crates/core/src/imap/client.rs +++ b/crates/core/src/imap/client.rs @@ -21,10 +21,11 @@ use crate::error::code::ErrorCode; use crate::error::BichonResult; use crate::imap::session::SessionStream; use crate::imap::stats::StatsWrapper; +use crate::imap::uidonly::{self, UidOnlyHandle, UidOnlyLimits}; +use crate::raise_error; use crate::utils::net::establish_tcp_connection_with_timeout; use crate::utils::net::establish_tls_connection; use crate::utils::tls::establish_tls_stream; -use crate::raise_error; use async_imap::Client as ImapClient; use async_imap::Session as ImapSession; use std::net::SocketAddr; @@ -87,6 +88,20 @@ impl Client { } } + /// Installs the UIDONLY protocol guard around the final transport. This is + /// called after STARTTLS has finished, but before authentication consumes + /// the client, so the same authenticated connection can be enabled later. + pub(crate) fn with_uidonly(self, limits: UidOnlyLimits) -> BichonResult<(Self, UidOnlyHandle)> { + let stream = self.inner.into_inner(); + let (stream, handle) = uidonly::wrap(stream, limits).map_err(|_| { + raise_error!( + "Invalid UIDONLY transport limits".into(), + ErrorCode::InvalidParameter + ) + })?; + Ok((Self::new(stream), handle)) + } + pub(crate) async fn login( self, username: &str, diff --git a/crates/core/src/imap/manager.rs b/crates/core/src/imap/manager.rs index e048edd0..64cec5fc 100644 --- a/crates/core/src/imap/manager.rs +++ b/crates/core/src/imap/manager.rs @@ -24,6 +24,7 @@ use crate::imap::capabilities::{capability_to_string, check_capabilities, fetch_ use crate::imap::client::Client; use crate::imap::oauth2::OAuth2; use crate::imap::session::SessionStream; +use crate::imap::uidonly::{UidOnlyHandle, UidOnlyLimits}; use crate::oauth2::token::OAuth2AccessToken; use crate::{bichon_version, decrypt, raise_error}; use async_imap::Session; @@ -31,6 +32,12 @@ use tracing::{error, warn}; pub struct ImapConnectionManager; +pub(crate) struct UidOnlyConnection { + pub session: Session>, + pub handle: UidOnlyHandle, + pub capabilities: Vec, +} + impl ImapConnectionManager { async fn create_client(account: &AccountModel) -> BichonResult { assert_eq!(account.account_type, AccountType::IMAP); @@ -93,13 +100,11 @@ impl ImapConnectionManager { } } - pub async fn build(account_id: u64) -> BichonResult>> { - let account = AccountModel::get(account_id)?; + async fn create_client_with_retry(account: &AccountModel) -> BichonResult { let account_email = account.email.clone(); - let mut client = None; for attempt in 0..3u32 { - match Self::create_client(&account).await { + match Self::create_client(account).await { Ok(c) => { client = Some(c); break; @@ -123,14 +128,19 @@ impl ImapConnectionManager { } } - let client = client.ok_or_else(|| { + client.ok_or_else(|| { raise_error!( format!("Failed to create IMAP {}'s client after 3 attempts", account_email), ErrorCode::NetworkError ) - })?; + }) + } - let mut session = match Self::authenticate(client, &account).await { + async fn initialize_session( + account: &AccountModel, + client: Client, + ) -> BichonResult<(Session>, Vec)> { + let mut session = match Self::authenticate(client, account).await { Ok(session) => session, Err(error) => { error!("Failed to authenticate IMAP session: {:#?}", error); @@ -138,10 +148,10 @@ impl ImapConnectionManager { } }; - match fetch_capabilities(&mut session).await { + let to_save = match fetch_capabilities(&mut session).await { Ok(capabilities) => { let to_save: Vec = capabilities.iter().map(capability_to_string).collect(); - AccountModel::update_capabilities(account_id, to_save)?; + AccountModel::update_capabilities(account.id, to_save.clone())?; if let Err(error) = check_capabilities(&capabilities) { error!("Failed to check IMAP capabilities: {:#?}", error); return Err(error); @@ -159,13 +169,38 @@ impl ImapConnectionManager { warn!("IMAP ID command failed (ignored): {:#?}", e); } } + to_save } Err(error) => { error!("Failed to fetch IMAP capabilities: {:#?}", error); return Err(error); } - } + }; + + Ok((session, to_save)) + } + pub async fn build(account_id: u64) -> BichonResult>> { + let account = AccountModel::get(account_id)?; + let client = Self::create_client_with_retry(&account).await?; + let (session, _) = Self::initialize_session(&account, client).await?; Ok(session) } + + /// Builds one capability-probed connection with the UIDONLY guard already + /// installed. The returned session is still in ordinary mode until an + /// exact `ENABLE UIDONLY` exchange succeeds. + pub(crate) async fn build_uidonly( + account: &AccountModel, + limits: UidOnlyLimits, + ) -> BichonResult { + let client = Self::create_client_with_retry(account).await?; + let (client, handle) = client.with_uidonly(limits)?; + let (session, capabilities) = Self::initialize_session(account, client).await?; + Ok(UidOnlyConnection { + session, + handle, + capabilities, + }) + } } diff --git a/crates/core/src/imap/mock_server.rs b/crates/core/src/imap/mock_server.rs index c37c3455..5cd5e141 100644 --- a/crates/core/src/imap/mock_server.rs +++ b/crates/core/src/imap/mock_server.rs @@ -38,7 +38,7 @@ //! ``` use std::net::SocketAddr; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{TcpListener, TcpStream}; @@ -47,6 +47,7 @@ type Response = Vec; pub struct MockImapServer { greeting: Vec, script: Vec<(String, Response)>, + commands: Arc>>, } impl MockImapServer { @@ -54,6 +55,7 @@ impl MockImapServer { Self { greeting: b"* OK Mock IMAP server ready\r\n".to_vec(), script: Vec::new(), + commands: Arc::new(Mutex::new(Vec::new())), } } @@ -77,6 +79,7 @@ impl MockImapServer { let addr = listener.local_addr().expect("local_addr"); let server = Arc::new(self); + let commands = Arc::clone(&server.commands); tokio::spawn(async move { loop { @@ -92,7 +95,7 @@ impl MockImapServer { } }); - MockImapServerHandle { addr } + MockImapServerHandle { addr, commands } } async fn handle_connection(&self, mut stream: TcpStream) { @@ -112,6 +115,10 @@ impl MockImapServer { Ok(_) => {} Err(_) => break, } + self.commands + .lock() + .expect("commands poisoned") + .push(line.trim_end_matches(['\r', '\n']).to_string()); let tag = extract_tag(&line).unwrap_or("A0"); let matched = self.find_match(&line); @@ -151,6 +158,7 @@ impl Default for MockImapServer { /// is dropped. pub struct MockImapServerHandle { addr: SocketAddr, + commands: Arc>>, } impl MockImapServerHandle { @@ -161,6 +169,10 @@ impl MockImapServerHandle { pub fn port(&self) -> u16 { self.addr.port() } + + pub fn commands(&self) -> Vec { + self.commands.lock().expect("commands poisoned").clone() + } } fn extract_tag(line: &str) -> Option<&str> { diff --git a/crates/core/src/imap/mod.rs b/crates/core/src/imap/mod.rs index 161c760c..3d623f09 100644 --- a/crates/core/src/imap/mod.rs +++ b/crates/core/src/imap/mod.rs @@ -27,6 +27,7 @@ pub mod stats; #[cfg(test)] mod tests; pub(crate) mod uidonly; +pub(crate) mod uidonly_acquisition; #[cfg(test)] mod uidonly_tests; #[cfg(test)] diff --git a/crates/core/src/imap/uidonly_acquisition.rs b/crates/core/src/imap/uidonly_acquisition.rs new file mode 100644 index 00000000..a376c66e --- /dev/null +++ b/crates/core/src/imap/uidonly_acquisition.rs @@ -0,0 +1,1823 @@ +// +// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) +// +// This file is part of the Bichon Email Archiving Project +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +//! Streaming, fail-closed acquisition for RFC 9586 UIDONLY mailboxes. +//! +//! There is deliberately no secondary ledger. The canonical envelope plus +//! verified exact-raw blob is the per-UID receipt. Transient disconnects retry +//! the same operation after re-proving the fixed mailbox epoch; a failed run +//! leaves the mailbox checkpoint unchanged. + +use crate::account::entity::{AuthType, Encryption}; +use crate::account::migration::AccountModel; +use crate::cache::imap::mailbox::MailBox; +use crate::envelope::extractor::{ + project_uidonly_messages, verify_uidonly_projections, UidOnlyMessage, + UIDONLY_PROJECTION_BATCH_BYTES, UIDONLY_PROJECTION_BATCH_MESSAGES, +}; +use crate::error::code::ErrorCode; +use crate::error::BichonResult; +use crate::imap::executor::DEFAULT_MAX_EMAIL_SIZE; +use crate::imap::manager::ImapConnectionManager; +use crate::imap::session::SessionStream; +use crate::imap::uidonly::{exact_args, inventory_args, UidOnlyHandle, UidOnlyLimits}; +use crate::raise_error; +use async_imap::Session; +use futures::{FutureExt, TryStreamExt}; +use std::future::Future; +use std::panic::AssertUnwindSafe; +use std::time::Duration; +use tokio_util::sync::CancellationToken; + +const UIDONLY_DEFAULT_PAGE_SIZE: u32 = 1_000; +const MAX_UIDONLY_RECONNECTS: u32 = 3; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct AcquisitionLimits { + pub max_literal_bytes: u64, + pub max_operation_runtime: Duration, + pub page_size: u32, +} + +impl AcquisitionLimits { + pub(crate) fn bounded(max_literal_bytes: u64) -> Self { + Self { + max_literal_bytes, + // Bound each network or durable-storage operation, not the whole + // archive: a valid large mailbox may need to run for days. + max_operation_runtime: Duration::from_secs(10 * 60), + page_size: 1_000, + } + } + + fn validate(self) -> BichonResult { + if self.max_literal_bytes == 0 + || self.max_operation_runtime.is_zero() + || self.page_size == 0 + { + return Err(raise_error!( + "UIDONLY acquisition limits must be nonzero".into(), + ErrorCode::InvalidParameter + )); + } + Ok(self) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct MailboxSnapshot { + pub exists: u32, + pub uid_validity: u32, + pub uid_next: u32, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct InventoryItem { + pub uid: u32, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct AcquisitionProgress { + pub planned: u64, + pub resolved: u64, + pub downloaded: u64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct AcquisitionReport { + pub uid_validity: u32, + pub uid_next: u32, + pub exists: u32, + pub checkpoint: Option, + pub inventoried: u64, + pub archived: u64, +} + +#[allow(async_fn_in_trait)] +pub(crate) trait UidOnlyTransport { + async fn snapshot(&mut self, mailbox: &str) -> BichonResult; + + async fn inventory_page( + &mut self, + cursor: u32, + high: u32, + page_size: u32, + ) -> BichonResult>; + + /// Fetch exactly one full message. `literal_budget` is a pre-read ceiling, + /// not a post-read accounting hint. + async fn fetch_exact(&mut self, uid: u32, literal_budget: u64) -> BichonResult; + + async fn reconnect(&mut self, _page_size: u32) -> BichonResult<()> { + Err(raise_error!( + "UIDONLY transport cannot reconnect".into(), + ErrorCode::NetworkError + )) + } +} + +#[allow(async_fn_in_trait)] +pub(crate) trait CanonicalArchive { + fn resume_after(&self) -> Option { + None + } + + fn begin_epoch(&mut self, _uid_validity: u32) -> BichonResult<()> { + Ok(()) + } + + async fn verify_many(&mut self, uids: &[u32]) -> BichonResult>; + + /// A successful result means every stored message has passed the durable + /// raw readback and final envelope-marker commit barrier. + async fn project_many(&mut self, messages: Vec) -> BichonResult<()>; +} + +async fn bounded(future: F, runtime: Duration, token: &CancellationToken) -> BichonResult +where + F: Future>, +{ + tokio::select! { + _ = token.cancelled() => Err(raise_error!( + "UIDONLY acquisition cancelled".into(), + ErrorCode::InternalError + )), + result = tokio::time::timeout(runtime, future) => result.map_err(|_| raise_error!( + "UIDONLY operation runtime ceiling exceeded".into(), + ErrorCode::RequestTimeout + ))?, + } +} + +fn retryable_transport_error(code: ErrorCode) -> bool { + matches!( + code, + ErrorCode::NetworkError | ErrorCode::ConnectionTimeout | ErrorCode::RequestTimeout + ) +} + +async fn recover_transport( + transport: &mut T, + mailbox: &str, + initial: MailboxSnapshot, + limits: AcquisitionLimits, + token: &CancellationToken, + reconnects: &mut u32, +) -> BichonResult<()> { + let mut last_error = raise_error!( + "UIDONLY transport recovery failed".into(), + ErrorCode::NetworkError + ); + while *reconnects < MAX_UIDONLY_RECONNECTS { + *reconnects += 1; + let delay = if cfg!(test) { + Duration::ZERO + } else { + Duration::from_secs(1 << (*reconnects - 1)) + }; + let reconnected = bounded( + async { + tokio::time::sleep(delay).await; + transport.reconnect(limits.page_size).await + }, + limits.max_operation_runtime, + token, + ) + .await; + if let Err(error) = reconnected { + if retryable_transport_error(error.code()) { + last_error = error; + continue; + } + return Err(error); + } + + match bounded( + transport.snapshot(mailbox), + limits.max_operation_runtime, + token, + ) + .await + { + Ok(snapshot) + if snapshot.uid_validity == initial.uid_validity + && snapshot.uid_next >= initial.uid_next => + { + return Ok(()) + } + Ok(_) => { + return Err(raise_error!( + "UIDONLY mailbox epoch changed while reconnecting".into(), + ErrorCode::Incompatible + )) + } + Err(error) if retryable_transport_error(error.code()) => last_error = error, + Err(error) => return Err(error), + } + } + Err(last_error) +} + +async fn inventory_with_reconnect( + transport: &mut T, + mailbox: &str, + initial: MailboxSnapshot, + cursor: u32, + high: u32, + limits: AcquisitionLimits, + token: &CancellationToken, +) -> BichonResult> { + let mut reconnects = 0; + loop { + match bounded( + transport.inventory_page(cursor, high, limits.page_size), + limits.max_operation_runtime, + token, + ) + .await + { + Err(error) if retryable_transport_error(error.code()) => { + recover_transport(transport, mailbox, initial, limits, token, &mut reconnects) + .await?; + } + result => return result, + } + } +} + +async fn fetch_with_reconnect( + transport: &mut T, + mailbox: &str, + initial: MailboxSnapshot, + uid: u32, + limits: AcquisitionLimits, + token: &CancellationToken, +) -> BichonResult { + let mut reconnects = 0; + loop { + match bounded( + transport.fetch_exact(uid, limits.max_literal_bytes), + limits.max_operation_runtime, + token, + ) + .await + { + Err(error) if retryable_transport_error(error.code()) => { + recover_transport(transport, mailbox, initial, limits, token, &mut reconnects) + .await?; + } + result => return result, + } + } +} + +async fn snapshot_with_reconnect( + transport: &mut T, + mailbox: &str, + initial: MailboxSnapshot, + limits: AcquisitionLimits, + token: &CancellationToken, +) -> BichonResult { + let mut reconnects = 0; + loop { + match bounded( + transport.snapshot(mailbox), + limits.max_operation_runtime, + token, + ) + .await + { + Err(error) if retryable_transport_error(error.code()) => { + recover_transport(transport, mailbox, initial, limits, token, &mut reconnects) + .await?; + } + result => return result, + } + } +} + +async fn flush_projection_batch( + archive: &mut A, + pending: &mut Vec, + limits: AcquisitionLimits, + token: &CancellationToken, +) -> BichonResult { + if pending.is_empty() { + return Ok(0); + } + let count = pending.len() as u64; + bounded( + archive.project_many(std::mem::take(pending)), + limits.max_operation_runtime, + token, + ) + .await?; + Ok(count) +} + +/// Reconciles one immutable UID range. Any ambiguity is an error, so callers +/// must persist `checkpoint` only from a returned report. +pub(crate) async fn run_acquisition( + transport: &mut T, + archive: &mut A, + mailbox: &str, + expected_uid_validity: Option, + limits: AcquisitionLimits, + token: CancellationToken, + mut progress: P, +) -> BichonResult +where + T: UidOnlyTransport, + A: CanonicalArchive, + P: FnMut(AcquisitionProgress) -> BichonResult<()>, +{ + let limits = limits.validate()?; + let snapshot = bounded( + transport.snapshot(mailbox), + limits.max_operation_runtime, + &token, + ) + .await?; + if snapshot.uid_validity == 0 || snapshot.uid_next == 0 { + return Err(raise_error!( + "UIDONLY EXAMINE omitted a valid UIDVALIDITY or UIDNEXT".into(), + ErrorCode::ImapUnexpectedResult + )); + } + if expected_uid_validity.is_some_and(|expected| expected != snapshot.uid_validity) { + return Err(raise_error!( + "UIDVALIDITY changed between mailbox discovery and UIDONLY EXAMINE".into(), + ErrorCode::Incompatible + )); + } + archive.begin_epoch(snapshot.uid_validity)?; + let planned = u64::from(snapshot.exists); + let high = snapshot.uid_next - 1; + let resume_after = archive.resume_after(); + if resume_after.is_some_and(|checkpoint| checkpoint > high) { + return Err(raise_error!( + "UIDONLY UIDNEXT moved behind the proven checkpoint".into(), + ErrorCode::Incompatible + )); + } + let full_snapshot = resume_after.is_none(); + let mut cursor = resume_after.map_or(1, |uid| uid.saturating_add(1)); + let mut inventoried = 0_u64; + let mut archived = 0_u64; + let mut downloaded = 0_u64; + let mut pending = Vec::with_capacity(UIDONLY_PROJECTION_BATCH_MESSAGES); + let mut pending_bytes = 0_u64; + + progress(AcquisitionProgress { + planned, + resolved: 0, + downloaded: 0, + })?; + + while high > 0 && cursor <= high { + let page = + inventory_with_reconnect(transport, mailbox, snapshot, cursor, high, limits, &token) + .await?; + if page.len() > limits.page_size as usize { + return Err(raise_error!( + "UIDONLY inventory page exceeded its requested bound".into(), + ErrorCode::ImapUnexpectedResult + )); + } + if page.is_empty() { + break; + } + + let mut previous = cursor - 1; + for item in &page { + if token.is_cancelled() { + return Err(raise_error!( + "UIDONLY acquisition cancelled".into(), + ErrorCode::InternalError + )); + } + if item.uid < cursor || item.uid > high || item.uid <= previous { + return Err(raise_error!( + "UIDONLY inventory was duplicate, unordered, or outside the fixed range".into(), + ErrorCode::ImapUnexpectedResult + )); + } + previous = item.uid; + inventoried = inventoried.checked_add(1).ok_or_else(|| { + raise_error!( + "UIDONLY inventory count overflow".into(), + ErrorCode::PayloadTooLarge + ) + })?; + } + let uids = page.iter().map(|item| item.uid).collect::>(); + let verified = bounded( + archive.verify_many(&uids), + limits.max_operation_runtime, + &token, + ) + .await?; + if verified.len() != page.len() { + return Err(raise_error!( + "UIDONLY receipt lookup returned the wrong result count".into(), + ErrorCode::InternalError + )); + } + + for (item, is_verified) in page.into_iter().zip(verified) { + if is_verified { + archived += 1; + continue; + } + + let message = + fetch_with_reconnect(transport, mailbox, snapshot, item.uid, limits, &token) + .await?; + if message.uid != item.uid { + return Err(raise_error!( + "UIDONLY exact fetch returned the wrong UID".into(), + ErrorCode::ImapUnexpectedResult + )); + } + let actual = message.body.len() as u64; + if actual > limits.max_literal_bytes { + return Err(raise_error!( + "UIDONLY exact body exceeded its pre-read budget".into(), + ErrorCode::PayloadTooLarge + )); + } + if !pending.is_empty() + && (pending.len() >= UIDONLY_PROJECTION_BATCH_MESSAGES + || pending_bytes.saturating_add(actual) > UIDONLY_PROJECTION_BATCH_BYTES as u64) + { + let stored = flush_projection_batch(archive, &mut pending, limits, &token).await?; + archived += stored; + downloaded += stored; + pending_bytes = 0; + } + pending_bytes += actual; + pending.push(message); + } + + let stored = flush_projection_batch(archive, &mut pending, limits, &token).await?; + archived += stored; + downloaded += stored; + pending_bytes = 0; + + cursor = previous.checked_add(1).unwrap_or(u32::MAX); + progress(AcquisitionProgress { + planned, + resolved: inventoried, + downloaded, + })?; + if previous == u32::MAX { + break; + } + } + + if (full_snapshot && inventoried != planned) || archived != inventoried { + return Err(raise_error!( + "UIDONLY inventory did not reconcile the EXAMINE message count".into(), + ErrorCode::ImapUnexpectedResult + )); + } + let final_snapshot = + snapshot_with_reconnect(transport, mailbox, snapshot, limits, &token).await?; + if final_snapshot.uid_validity != snapshot.uid_validity + || final_snapshot.uid_next < snapshot.uid_next + { + return Err(raise_error!( + "UIDONLY mailbox epoch changed during acquisition".into(), + ErrorCode::Incompatible + )); + } + Ok(AcquisitionReport { + uid_validity: snapshot.uid_validity, + uid_next: snapshot.uid_next, + exists: snapshot.exists, + checkpoint: (high > 0).then_some(high), + inventoried, + archived, + }) +} + +pub(crate) enum AcquisitionRoute { + Acquired { + report: AcquisitionReport, + source_scope: String, + }, + Legacy(Session>), +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum CapabilityRoute { + Legacy, + UidOnly { message_limit: u32 }, +} + +fn is_capability(value: &str, expected: &str) -> bool { + value.trim().eq_ignore_ascii_case(expected) +} + +fn is_message_limit_marker(value: &str) -> bool { + value + .trim() + .get(.."MESSAGELIMIT".len()) + .is_some_and(|head| head.eq_ignore_ascii_case("MESSAGELIMIT")) +} + +fn classify_capabilities(capabilities: &[String]) -> BichonResult { + let uidonly = capabilities + .iter() + .filter(|value| is_capability(value, "UIDONLY")) + .count(); + let partial = capabilities + .iter() + .filter(|value| is_capability(value, "PARTIAL")) + .count(); + let limits: Vec<_> = capabilities + .iter() + .filter(|value| is_message_limit_marker(value)) + .collect(); + + // PARTIAL is a standalone RFC 9394 extension. It is not by itself + // evidence that ordinary mailbox views are limited. + if uidonly == 0 && limits.is_empty() { + return Ok(CapabilityRoute::Legacy); + } + if uidonly != 1 || partial != 1 || limits.len() != 1 { + return Err(raise_error!( + "Server advertised an incomplete or ambiguous UIDONLY capability set".into(), + ErrorCode::Incompatible + )); + } + let value = limits[0].trim(); + let prefix = "MESSAGELIMIT="; + let number = value + .get(prefix.len()..) + .filter(|_| { + value + .get(..prefix.len()) + .is_some_and(|head| head.eq_ignore_ascii_case(prefix)) + }) + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .ok_or_else(|| { + raise_error!( + "Server advertised an invalid UIDONLY MESSAGELIMIT".into(), + ErrorCode::Incompatible + ) + })?; + Ok(CapabilityRoute::UidOnly { + message_limit: number, + }) +} + +fn require_uidonly_opt_in( + account: &AccountModel, + capability_route: CapabilityRoute, +) -> BichonResult<()> { + if matches!(capability_route, CapabilityRoute::UidOnly { .. }) && !account.uidonly_enabled { + return Err(raise_error!( + "Server requires UIDONLY acquisition; enable uidonly_enabled for this account after reviewing the storage transition".into(), + ErrorCode::Incompatible + )); + } + Ok(()) +} + +/// Cached capabilities only decide whether legacy reconciliation may be +/// bypassed. The acquisition connection always reclassifies fresh capabilities. +pub(crate) fn account_requires_uidonly(account: &AccountModel) -> bool { + account.capabilities.as_ref().is_some_and(|capabilities| { + capabilities + .iter() + .any(|value| is_capability(value, "UIDONLY") || is_message_limit_marker(value)) + }) +} + +pub(crate) fn mailbox_has_uidonly_proof(account: &AccountModel, mailbox: &MailBox) -> bool { + source_scope(account) + .ok() + .as_deref() + .is_some_and(|scope| mailbox.uidonly_source_scope.as_deref() == Some(scope)) +} + +fn source_scope(account: &AccountModel) -> BichonResult { + let imap = account.imap.as_ref().ok_or_else(|| { + raise_error!( + "UIDONLY account has no IMAP configuration".into(), + ErrorCode::MissingConfiguration + ) + })?; + let host = imap.host.trim().trim_end_matches('.').to_ascii_lowercase(); + let principal = account.login_name.as_ref().unwrap_or(&account.email); + if host.is_empty() || principal.is_empty() || imap.port == 0 { + return Err(raise_error!( + "UIDONLY source identity is incomplete".into(), + ErrorCode::MissingConfiguration + )); + } + let mut hasher = blake3::Hasher::new(); + hasher.update(b"bichon-uidonly-source-v2\0"); + for field in [host.as_bytes(), principal.as_bytes()] { + hasher.update(&(field.len() as u64).to_be_bytes()); + hasher.update(field); + } + hasher.update(&imap.port.to_be_bytes()); + Ok(hasher.finalize().to_hex().to_string()) +} + +fn connection_scope(account: &AccountModel) -> BichonResult { + let imap = account.imap.as_ref().ok_or_else(|| { + raise_error!( + "UIDONLY account has no IMAP configuration".into(), + ErrorCode::MissingConfiguration + ) + })?; + let encryption = match imap.encryption { + Encryption::Ssl => 1_u8, + Encryption::StartTls => 2, + Encryption::None => 3, + }; + let mut hasher = blake3::Hasher::new(); + hasher.update(b"bichon-uidonly-connection-v1\0"); + hasher.update(source_scope(account)?.as_bytes()); + hasher.update(&[encryption, u8::from(account.use_dangerous)]); + match imap.use_proxy { + Some(proxy) => { + hasher.update(&[1]); + hasher.update(&proxy.to_be_bytes()); + } + None => { + hasher.update(&[0]); + } + } + hasher.update(&[match imap.auth.auth_type { + AuthType::Password => 1, + AuthType::OAuth2 => 2, + }]); + Ok(hasher.finalize().to_hex().to_string()) +} + +fn protocol_limits(max_literal_bytes: u64) -> BichonResult { + if max_literal_bytes == 0 || max_literal_bytes > u64::from(u32::MAX) { + return Err(raise_error!( + "UIDONLY message-size limit is outside the supported range".into(), + ErrorCode::InvalidParameter + )); + } + let literal = usize::try_from(max_literal_bytes).map_err(|_| { + raise_error!( + "UIDONLY message-size limit is not representable".into(), + ErrorCode::InvalidParameter + ) + })?; + let response = literal.checked_add(128 * 1024).ok_or_else(|| { + raise_error!( + "UIDONLY response-size limit overflow".into(), + ErrorCode::InvalidParameter + ) + })?; + let command = response.checked_add(128 * 1024).ok_or_else(|| { + raise_error!( + "UIDONLY command-size limit overflow".into(), + ErrorCode::InvalidParameter + ) + })?; + Ok(UidOnlyLimits { + max_control_line_bytes: 64 * 1024, + max_literal_bytes: literal, + max_response_bytes: response, + max_command_literal_bytes: literal, + max_command_response_bytes: command, + // A 1,000-message inventory page plus bounded unsolicited mailbox + // updates must still fit without making the response count unbounded. + max_command_responses: 2_048, + max_command_runtime: Duration::from_secs(5 * 60), + }) +} + +fn protocol_probe_literal(max_literal_bytes: u64) -> u64 { + let representable = u64::from(u32::MAX).min(usize::MAX.saturating_sub(256 * 1024) as u64); + max_literal_bytes.clamp(1, representable) +} + +fn imap_error_code(error: &async_imap::error::Error) -> ErrorCode { + match error { + async_imap::error::Error::ConnectionLost => ErrorCode::NetworkError, + async_imap::error::Error::Io(error) + if matches!( + error.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::NotConnected + | std::io::ErrorKind::UnexpectedEof + | std::io::ErrorKind::TimedOut + | std::io::ErrorKind::WriteZero + ) => + { + ErrorCode::NetworkError + } + _ => ErrorCode::ImapUnexpectedResult, + } +} + +struct SessionUidOnlyTransport { + session: Session>, + handle: UidOnlyHandle, + account: AccountModel, + limits: UidOnlyLimits, + connection_scope: String, +} + +impl SessionUidOnlyTransport { + fn healthy_handle(handle: &UidOnlyHandle) -> BichonResult<()> { + if let Some(reason) = handle.poison_reason() { + return Err(raise_error!(reason, ErrorCode::ImapUnexpectedResult)); + } + handle.ensure_active().map_err(|_| { + raise_error!( + "UIDONLY transport is not active".into(), + ErrorCode::ImapUnexpectedResult + ) + }) + } + + fn healthy(&self) -> BichonResult<()> { + Self::healthy_handle(&self.handle) + } + + async fn enable( + session: &mut Session>, + handle: &UidOnlyHandle, + ) -> BichonResult<()> { + session + .run_command_and_check_ok("ENABLE UIDONLY") + .await + .map_err(|error| { + raise_error!( + "Server did not enable UIDONLY".into(), + imap_error_code(&error) + ) + })?; + Self::healthy_handle(handle) + } + + async fn collect_fetches( + &mut self, + set: String, + query: impl AsRef, + ) -> BichonResult> { + let query = query.as_ref().to_string(); + let operation = async { + let stream = self.session.uid_fetch(set, query).await.map_err(|error| { + raise_error!( + "UIDONLY fetch command failed".into(), + imap_error_code(&error) + ) + })?; + stream.try_collect::>().await.map_err(|error| { + raise_error!( + "UIDONLY fetch response failed".into(), + imap_error_code(&error) + ) + }) + }; + let result = AssertUnwindSafe(operation) + .catch_unwind() + .await + .map_err(|_| { + raise_error!( + "UIDONLY parser rejected a malformed response".into(), + ErrorCode::ImapUnexpectedResult + ) + })??; + self.healthy()?; + Ok(result) + } +} + +impl UidOnlyTransport for SessionUidOnlyTransport { + async fn snapshot(&mut self, mailbox: &str) -> BichonResult { + self.healthy()?; + let mailbox = AssertUnwindSafe(self.session.examine(mailbox)) + .catch_unwind() + .await + .map_err(|_| { + raise_error!( + "UIDONLY parser rejected EXAMINE".into(), + ErrorCode::ImapUnexpectedResult + ) + })? + .map_err(|error| { + raise_error!("UIDONLY EXAMINE failed".into(), imap_error_code(&error)) + })?; + self.healthy()?; + Ok(MailboxSnapshot { + exists: mailbox.exists, + uid_validity: mailbox.uid_validity.ok_or_else(|| { + raise_error!( + "UIDONLY EXAMINE omitted UIDVALIDITY".into(), + ErrorCode::ImapUnexpectedResult + ) + })?, + uid_next: mailbox.uid_next.ok_or_else(|| { + raise_error!( + "UIDONLY EXAMINE omitted UIDNEXT".into(), + ErrorCode::ImapUnexpectedResult + ) + })?, + }) + } + + async fn inventory_page( + &mut self, + cursor: u32, + high: u32, + page_size: u32, + ) -> BichonResult> { + let (set, query) = inventory_args(cursor, high, page_size).map_err(|_| { + raise_error!( + "Invalid UIDONLY inventory bounds".into(), + ErrorCode::InvalidParameter + ) + })?; + self.collect_fetches(set, query) + .await? + .into_iter() + .map(|fetch| { + let uid = fetch.uid.ok_or_else(|| { + raise_error!( + "UIDONLY inventory omitted UID".into(), + ErrorCode::ImapUnexpectedResult + ) + })?; + if fetch.message != uid { + return Err(raise_error!( + "UIDONLY inventory leading UID did not match UID data item".into(), + ErrorCode::ImapUnexpectedResult + )); + } + fetch.size.ok_or_else(|| { + raise_error!( + "UIDONLY inventory omitted RFC822.SIZE".into(), + ErrorCode::ImapUnexpectedResult + ) + })?; + Ok(InventoryItem { uid }) + }) + .collect() + } + + async fn fetch_exact(&mut self, uid: u32, literal_budget: u64) -> BichonResult { + let limit = usize::try_from(literal_budget).map_err(|_| { + raise_error!( + "UIDONLY literal budget is not representable".into(), + ErrorCode::InvalidParameter + ) + })?; + self.handle + .arm_next_fetch_literal_limit(limit) + .map_err(|_| { + raise_error!( + "UIDONLY exact fetch could not arm its pre-read limit".into(), + ErrorCode::ImapUnexpectedResult + ) + })?; + let before = self.handle.literal_bytes_received(); + let (set, query) = exact_args(uid).map_err(|_| { + raise_error!( + "Invalid UIDONLY exact UID".into(), + ErrorCode::InvalidParameter + ) + })?; + let mut fetched = self.collect_fetches(set, query).await?; + if fetched.len() != 1 { + return Err(raise_error!( + "UIDONLY exact fetch did not return exactly one message".into(), + ErrorCode::ImapUnexpectedResult + )); + } + let fetch = fetched.pop().expect("length checked"); + if fetch.message != uid || fetch.uid != Some(uid) || fetch.size.is_none() { + return Err(raise_error!( + "UIDONLY exact fetch returned mismatched metadata".into(), + ErrorCode::ImapUnexpectedResult + )); + } + let raw = fetch.body().ok_or_else(|| { + raise_error!( + "UIDONLY exact fetch omitted its full literal body".into(), + ErrorCode::ImapUnexpectedResult + ) + })?; + let after = self.handle.literal_bytes_received(); + if after.checked_sub(before) != Some(raw.len() as u64) { + return Err(raise_error!( + "UIDONLY exact body did not match literal accounting".into(), + ErrorCode::ImapUnexpectedResult + )); + } + Ok(UidOnlyMessage { + uid, + body: raw.to_vec(), + }) + } + + async fn reconnect(&mut self, page_size: u32) -> BichonResult<()> { + let current = AccountModel::get(self.account.id)?; + if connection_scope(¤t)? != self.connection_scope { + return Err(raise_error!( + "UIDONLY account connection changed while reconnecting".into(), + ErrorCode::Incompatible + )); + } + let connection = + ImapConnectionManager::build_uidonly(&self.account, self.limits.clone()).await?; + let CapabilityRoute::UidOnly { message_limit } = + classify_capabilities(&connection.capabilities)? + else { + return Err(raise_error!( + "UIDONLY reconnect lost required capabilities".into(), + ErrorCode::Incompatible + )); + }; + if message_limit < page_size { + return Err(raise_error!( + "UIDONLY reconnect reduced MESSAGELIMIT below the fixed page size".into(), + ErrorCode::Incompatible + )); + } + let mut session = connection.session; + Self::enable(&mut session, &connection.handle).await?; + self.session = session; + self.handle = connection.handle; + Ok(()) + } +} + +struct BichonCanonicalArchive { + account_id: u64, + mailbox_id: u64, + source_scope: String, + uid_validity: Option, + resume_after: Option, +} + +impl CanonicalArchive for BichonCanonicalArchive { + fn resume_after(&self) -> Option { + self.resume_after + } + + fn begin_epoch(&mut self, uid_validity: u32) -> BichonResult<()> { + if self + .uid_validity + .is_some_and(|existing| existing != uid_validity) + { + return Err(raise_error!( + "UIDONLY archive epoch changed".into(), + ErrorCode::Incompatible + )); + } + self.uid_validity = Some(uid_validity); + Ok(()) + } + + async fn verify_many(&mut self, uids: &[u32]) -> BichonResult> { + verify_uidonly_projections( + self.account_id, + self.mailbox_id, + self.uid_validity.expect("begin_epoch is called first"), + uids, + &self.source_scope, + ) + } + + async fn project_many(&mut self, messages: Vec) -> BichonResult<()> { + project_uidonly_messages( + messages, + self.account_id, + self.mailbox_id, + self.uid_validity.expect("begin_epoch is called first"), + &self.source_scope, + ) + .await + } +} + +pub(crate) async fn connect_and_acquire_or_legacy

( + account: &AccountModel, + mailbox: &MailBox, + force_uidonly: bool, + token: CancellationToken, + progress: P, +) -> BichonResult +where + P: FnMut(AcquisitionProgress) -> BichonResult<()>, +{ + let configured_scope = source_scope(account)?; + let known_limited = force_uidonly + || account_requires_uidonly(account) + || mailbox.uidonly_source_scope.as_deref() == Some(&configured_scope); + if known_limited + && account + .archive_rules + .as_ref() + .is_some_and(|rules| rules.enabled) + { + return Err(raise_error!( + "UIDONLY acquisition does not yet support enabled archive rules".into(), + ErrorCode::Incompatible + )); + } + let max_literal = account + .max_email_size_bytes + .unwrap_or(DEFAULT_MAX_EMAIL_SIZE); + let wire_limits = protocol_limits(protocol_probe_literal(max_literal))?; + let connection = ImapConnectionManager::build_uidonly(account, wire_limits.clone()).await?; + let capability_route = classify_capabilities(&connection.capabilities)?; + require_uidonly_opt_in(account, capability_route)?; + if capability_route == CapabilityRoute::Legacy { + if known_limited { + return Err(raise_error!( + "Known limited server omitted UIDONLY capabilities on the acquisition connection" + .into(), + ErrorCode::Incompatible + )); + } + return Ok(AcquisitionRoute::Legacy(connection.session)); + } + let CapabilityRoute::UidOnly { message_limit } = capability_route else { + unreachable!("legacy returned above") + }; + let wire_limits = protocol_limits(max_literal)?; + if account + .archive_rules + .as_ref() + .is_some_and(|rules| rules.enabled) + { + return Err(raise_error!( + "UIDONLY acquisition does not yet support enabled archive rules".into(), + ErrorCode::Incompatible + )); + } + + let frozen_connection = connection_scope(account)?; + let mut transport = SessionUidOnlyTransport { + session: connection.session, + handle: connection.handle, + account: account.clone(), + limits: wire_limits, + connection_scope: frozen_connection.clone(), + }; + SessionUidOnlyTransport::enable(&mut transport.session, &transport.handle).await?; + + let frozen_scope = configured_scope; + let resume_after = (mailbox.uidonly_source_scope.as_deref() == Some(&frozen_scope)) + .then_some(mailbox.highest_uid) + .flatten(); + let mut archive = BichonCanonicalArchive { + account_id: account.id, + mailbox_id: mailbox.id, + source_scope: frozen_scope.clone(), + uid_validity: None, + resume_after, + }; + let mut limits = AcquisitionLimits::bounded(max_literal); + limits.page_size = UIDONLY_DEFAULT_PAGE_SIZE.min(message_limit).max(1); + let result = run_acquisition( + &mut transport, + &mut archive, + &mailbox.encoded_name(), + mailbox.uid_validity, + limits, + token, + progress, + ) + .await; + let report = match result { + Ok(report) => report, + Err(error) => return Err(error), + }; + + let current = AccountModel::get(account.id)?; + if source_scope(¤t)? != frozen_scope + || connection_scope(¤t)? != frozen_connection + || current + .archive_rules + .as_ref() + .is_some_and(|rules| rules.enabled) + { + return Err(raise_error!( + "UIDONLY account source, connection policy, or archive rules changed during acquisition" + .into(), + ErrorCode::Incompatible + )); + } + transport.session.logout().await.ok(); + Ok(AcquisitionRoute::Acquired { + report, + source_scope: frozen_scope, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::account::entity::{AuthConfig, ImapConfig}; + use crate::imap::client::Client; + use crate::imap::mock_server::{examine_response, MockImapServer, MockImapServerHandle}; + use std::collections::{BTreeMap, BTreeSet, VecDeque}; + + fn limits() -> AcquisitionLimits { + AcquisitionLimits { + max_literal_bytes: 64 * 1024, + max_operation_runtime: Duration::from_secs(20), + page_size: 2, + } + } + + #[test] + fn capability_routing_ignores_standalone_partial_and_rejects_partial_uidonly_sets() { + assert_eq!( + classify_capabilities(&["IMAP4rev1".into(), "PARTIAL".into()]).unwrap(), + CapabilityRoute::Legacy + ); + assert_eq!( + classify_capabilities(&[ + "IMAP4rev1".into(), + "uidonly".into(), + "partial".into(), + "messagelimit=10000".into(), + ]) + .unwrap(), + CapabilityRoute::UidOnly { + message_limit: 10_000 + } + ); + for capabilities in [ + vec!["UIDONLY".into(), "PARTIAL".into()], + vec!["MESSAGELIMIT=1000".into(), "PARTIAL".into()], + vec!["UIDONLY".into(), "PARTIAL".into(), "MESSAGELIMIT=0".into()], + ] { + assert!(classify_capabilities(&capabilities).is_err()); + } + let cached = |capability: &str| AccountModel { + capabilities: Some(vec![capability.into()]), + ..Default::default() + }; + assert!(!account_requires_uidonly(&cached("PARTIAL"))); + assert!(account_requires_uidonly(&cached("UIDONLY"))); + assert!(account_requires_uidonly(&cached("MESSAGELIMIT=10000"))); + } + + #[test] + fn uidonly_route_requires_explicit_account_opt_in() { + let route = CapabilityRoute::UidOnly { + message_limit: 10_000, + }; + let error = require_uidonly_opt_in(&AccountModel::default(), route).unwrap_err(); + assert_eq!(error.code(), ErrorCode::Incompatible); + assert!(error.to_string().contains("enable uidonly_enabled")); + + let enabled = AccountModel { + uidonly_enabled: true, + ..Default::default() + }; + assert!(require_uidonly_opt_in(&enabled, route).is_ok()); + assert!(require_uidonly_opt_in(&AccountModel::default(), CapabilityRoute::Legacy).is_ok()); + } + + #[test] + fn capability_probe_accepts_limits_that_only_uidonly_rejects() { + for configured in [0, u64::from(u32::MAX) + 1] { + assert!(protocol_limits(protocol_probe_literal(configured)).is_ok()); + assert!(protocol_limits(configured).is_err()); + } + } + + #[test] + fn source_identity_is_stable_across_connection_policy_changes() { + let account = AccountModel { + email: "alice@example.invalid".into(), + imap: Some(ImapConfig { + host: "IMAP.EXAMPLE.INVALID.".into(), + port: 993, + encryption: Encryption::Ssl, + auth: AuthConfig { + auth_type: AuthType::Password, + password: None, + }, + use_proxy: None, + }), + ..Default::default() + }; + let source = source_scope(&account).unwrap(); + let connection = connection_scope(&account).unwrap(); + for changed in [ + AccountModel { + use_dangerous: true, + ..account.clone() + }, + AccountModel { + imap: account.imap.clone().map(|mut imap| { + imap.use_proxy = Some(7); + imap + }), + ..account.clone() + }, + AccountModel { + imap: account.imap.clone().map(|mut imap| { + imap.auth.auth_type = AuthType::OAuth2; + imap + }), + ..account.clone() + }, + ] { + assert_eq!(source_scope(&changed).unwrap(), source); + assert_ne!(connection_scope(&changed).unwrap(), connection); + } + let mut other_principal = account; + other_principal.login_name = Some("other@example.invalid".into()); + assert_ne!(source_scope(&other_principal).unwrap(), source); + } + + struct FakeTransport { + snapshot: MailboxSnapshot, + pages: VecDeque>, + messages: BTreeMap>, + fetched: Vec, + expected_cursor: Option, + } + + impl FakeTransport { + fn sparse() -> Self { + let body = |uid| { + format!("From: sender@invalid\r\nMessage-ID: <{uid}@invalid>\r\n\r\nbody") + .into_bytes() + }; + Self { + snapshot: MailboxSnapshot { + exists: 3, + uid_validity: 77, + uid_next: 51, + }, + pages: VecDeque::from([ + vec![InventoryItem { uid: 2 }, InventoryItem { uid: 30 }], + vec![InventoryItem { uid: 50 }], + vec![], + ]), + messages: BTreeMap::from([(2, body(2)), (30, body(30)), (50, body(50))]), + fetched: Vec::new(), + expected_cursor: None, + } + } + } + + impl UidOnlyTransport for FakeTransport { + async fn snapshot(&mut self, _mailbox: &str) -> BichonResult { + Ok(self.snapshot) + } + + async fn inventory_page( + &mut self, + cursor: u32, + _high: u32, + _page_size: u32, + ) -> BichonResult> { + if let Some(expected) = self.expected_cursor.take() { + assert_eq!(cursor, expected); + } + Ok(self.pages.pop_front().unwrap_or_default()) + } + + async fn fetch_exact( + &mut self, + uid: u32, + literal_budget: u64, + ) -> BichonResult { + self.fetched.push(uid); + let raw = self.messages.get(&uid).cloned().ok_or_else(|| { + raise_error!( + "synthetic missing body".into(), + ErrorCode::ImapUnexpectedResult + ) + })?; + if raw.len() as u64 > literal_budget { + return Err(raise_error!( + "synthetic literal budget".into(), + ErrorCode::PayloadTooLarge + )); + } + Ok(UidOnlyMessage { uid, body: raw }) + } + } + + struct FlakyTransport { + inner: FakeTransport, + inventory_failures: u32, + fetch_failures: u32, + reconnects: u32, + inventory_attempts: u32, + fetch_attempts: Vec, + snapshot_after_reconnect: Option, + } + + impl FlakyTransport { + fn sparse() -> Self { + Self { + inner: FakeTransport::sparse(), + inventory_failures: 0, + fetch_failures: 0, + reconnects: 0, + inventory_attempts: 0, + fetch_attempts: Vec::new(), + snapshot_after_reconnect: None, + } + } + } + + impl UidOnlyTransport for FlakyTransport { + async fn snapshot(&mut self, mailbox: &str) -> BichonResult { + self.inner.snapshot(mailbox).await + } + + async fn inventory_page( + &mut self, + cursor: u32, + high: u32, + page_size: u32, + ) -> BichonResult> { + self.inventory_attempts += 1; + if self.inventory_failures > 0 { + self.inventory_failures -= 1; + return Err(raise_error!( + "synthetic disconnect".into(), + ErrorCode::NetworkError + )); + } + self.inner.inventory_page(cursor, high, page_size).await + } + + async fn fetch_exact( + &mut self, + uid: u32, + literal_budget: u64, + ) -> BichonResult { + self.fetch_attempts.push(uid); + if self.fetch_failures > 0 { + self.fetch_failures -= 1; + return Err(raise_error!( + "synthetic disconnect".into(), + ErrorCode::NetworkError + )); + } + self.inner.fetch_exact(uid, literal_budget).await + } + + async fn reconnect(&mut self, _page_size: u32) -> BichonResult<()> { + self.reconnects += 1; + if let Some(snapshot) = self.snapshot_after_reconnect.take() { + self.inner.snapshot = snapshot; + } + Ok(()) + } + } + + #[derive(Default)] + struct FakeArchive { + verified: BTreeSet, + projected: Vec, + fail_project: bool, + resume_after: Option, + } + + impl CanonicalArchive for FakeArchive { + fn resume_after(&self) -> Option { + self.resume_after + } + + async fn verify_many(&mut self, uids: &[u32]) -> BichonResult> { + Ok(uids.iter().map(|uid| self.verified.contains(uid)).collect()) + } + + async fn project_many(&mut self, messages: Vec) -> BichonResult<()> { + if self.fail_project { + return Err(raise_error!( + "synthetic projection failure".into(), + ErrorCode::InternalError + )); + } + for message in messages { + self.projected.push(message.uid); + self.verified.insert(message.uid); + } + Ok(()) + } + } + + async fn run_test( + transport: &mut T, + archive: &mut A, + ) -> BichonResult { + run_acquisition( + transport, + archive, + "synthetic", + Some(77), + limits(), + CancellationToken::new(), + |_| Ok(()), + ) + .await + } + + #[tokio::test] + async fn sparse_snapshot_streams_and_checkpoints_fixed_high() { + let mut transport = FakeTransport::sparse(); + let mut archive = FakeArchive::default(); + let report = run_test(&mut transport, &mut archive).await.unwrap(); + assert_eq!(report.checkpoint, Some(50)); + assert_eq!((report.inventoried, report.archived), (3, 3)); + assert_eq!(transport.fetched, [2, 30, 50]); + } + + #[tokio::test] + async fn reconnect_retries_the_same_inventory_cursor() { + let mut transport = FlakyTransport::sparse(); + transport.inventory_failures = 1; + let report = run_test(&mut transport, &mut FakeArchive::default()) + .await + .unwrap(); + + assert_eq!(report.checkpoint, Some(50)); + assert_eq!(transport.reconnects, 1); + assert_eq!(transport.inventory_attempts, 3); + } + + #[tokio::test] + async fn reconnect_retries_one_exact_uid_without_double_projection() { + let mut transport = FlakyTransport::sparse(); + transport.fetch_failures = 1; + let mut archive = FakeArchive::default(); + let report = run_test(&mut transport, &mut archive).await.unwrap(); + + assert_eq!(report.archived, 3); + assert_eq!(transport.reconnects, 1); + assert_eq!(transport.fetch_attempts, [2, 2, 30, 50]); + assert_eq!(archive.projected, [2, 30, 50]); + } + + #[tokio::test] + async fn reconnect_rejects_changed_uidvalidity_or_lower_uidnext() { + for changed in [ + MailboxSnapshot { + uid_validity: 78, + ..FakeTransport::sparse().snapshot + }, + MailboxSnapshot { + uid_next: 50, + ..FakeTransport::sparse().snapshot + }, + ] { + let mut transport = FlakyTransport::sparse(); + transport.inventory_failures = 1; + transport.snapshot_after_reconnect = Some(changed); + let error = run_test(&mut transport, &mut FakeArchive::default()) + .await + .unwrap_err(); + + assert_eq!(error.code(), ErrorCode::Incompatible); + assert!(transport.inner.fetched.is_empty()); + } + } + + #[tokio::test] + async fn persistent_disconnect_uses_only_three_reconnects() { + let mut transport = FlakyTransport::sparse(); + transport.inventory_failures = 4; + let error = run_test(&mut transport, &mut FakeArchive::default()) + .await + .unwrap_err(); + + assert_eq!(error.code(), ErrorCode::NetworkError); + assert_eq!(transport.reconnects, 3); + assert_eq!(transport.inventory_attempts, 4); + } + + #[tokio::test] + async fn proven_checkpoint_scans_only_new_uids() { + let mut transport = FakeTransport::sparse(); + transport.snapshot.exists = 4; + transport.snapshot.uid_next = 60; + transport.pages = VecDeque::from([vec![InventoryItem { uid: 55 }]]); + transport + .messages + .insert(55, b"Subject: new\r\n\r\nbody".to_vec()); + transport.expected_cursor = Some(51); + let report = run_test( + &mut transport, + &mut FakeArchive { + resume_after: Some(50), + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!((report.inventoried, report.archived), (1, 1)); + assert_eq!(report.checkpoint, Some(59)); + assert_eq!(transport.fetched, [55]); + } + + #[tokio::test] + async fn uidnext_behind_proven_checkpoint_fails_closed() { + let mut transport = FakeTransport::sparse(); + let error = run_test( + &mut transport, + &mut FakeArchive { + resume_after: Some(51), + ..Default::default() + }, + ) + .await + .unwrap_err(); + assert_eq!(error.code(), ErrorCode::Incompatible); + assert!(transport.fetched.is_empty()); + } + + #[tokio::test] + async fn restart_skips_verified_receipts_without_a_secondary_ledger() { + let mut transport = FakeTransport::sparse(); + let mut archive = FakeArchive { + verified: BTreeSet::from([2, 30]), + ..Default::default() + }; + let report = run_test(&mut transport, &mut archive).await.unwrap(); + assert_eq!(report.archived, 3); + assert_eq!(transport.fetched, [50]); + } + + #[tokio::test] + async fn short_inventory_never_returns_a_checkpoint() { + let mut transport = FakeTransport::sparse(); + transport.snapshot.exists = 4; + let error = run_test(&mut transport, &mut FakeArchive::default()) + .await + .unwrap_err(); + assert_eq!(error.code(), ErrorCode::ImapUnexpectedResult); + } + + #[tokio::test] + async fn cancellation_and_projection_failure_abort_the_run() { + let token = CancellationToken::new(); + let cancel = token.clone(); + let mut transport = FakeTransport::sparse(); + let error = run_acquisition( + &mut transport, + &mut FakeArchive::default(), + "synthetic", + Some(77), + limits(), + token, + move |progress| { + if progress.resolved > 0 { + cancel.cancel(); + } + Ok(()) + }, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("cancelled")); + + let mut transport = FakeTransport::sparse(); + let error = run_test( + &mut transport, + &mut FakeArchive { + fail_project: true, + ..Default::default() + }, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("synthetic projection failure")); + } + + struct LogicalMillion { + cursor: u32, + } + + impl UidOnlyTransport for LogicalMillion { + async fn snapshot(&mut self, _mailbox: &str) -> BichonResult { + Ok(MailboxSnapshot { + exists: 1_000_000, + uid_validity: 9, + uid_next: 1_000_001, + }) + } + + async fn inventory_page( + &mut self, + cursor: u32, + high: u32, + page_size: u32, + ) -> BichonResult> { + assert_eq!(cursor, self.cursor); + let end = high.min(cursor + page_size - 1); + self.cursor = end + 1; + Ok((cursor..=end).map(|uid| InventoryItem { uid }).collect()) + } + + async fn fetch_exact( + &mut self, + _uid: u32, + _literal_budget: u64, + ) -> BichonResult { + unreachable!("every logical receipt verifies") + } + } + + struct AllVerified; + + impl CanonicalArchive for AllVerified { + async fn verify_many(&mut self, uids: &[u32]) -> BichonResult> { + Ok(vec![true; uids.len()]) + } + + async fn project_many(&mut self, _messages: Vec) -> BichonResult<()> { + unreachable!("every logical receipt verifies") + } + } + + #[tokio::test] + async fn million_message_inventory_is_page_bounded() { + let mut transport = LogicalMillion { cursor: 1 }; + let mut bounded = limits(); + bounded.page_size = 1_000; + bounded.max_operation_runtime = Duration::from_secs(30); + let report = run_acquisition( + &mut transport, + &mut AllVerified, + "synthetic", + Some(9), + bounded, + CancellationToken::new(), + |_| Ok(()), + ) + .await + .unwrap(); + assert_eq!(report.inventoried, 1_000_000); + assert_eq!(report.checkpoint, Some(1_000_000)); + } + + fn inventory_response(entries: &[(u32, u32)]) -> Vec { + let mut response = + b"* 3 EXISTS\r\n* OK [UIDNEXT 10] unchanged\r\n* 9 UIDFETCH (FLAGS (\\Seen))\r\n" + .to_vec(); + for (uid, size) in entries { + response.extend_from_slice( + format!("* {uid} UIDFETCH (UID {uid} RFC822.SIZE {size})\r\n").as_bytes(), + ); + } + response.extend_from_slice(b"{TAG} OK inventory completed\r\n"); + response + } + + fn exact_response(uid: u32, reported_size: u32, raw: &[u8]) -> Vec { + let mut response = format!( + "* {uid} UIDFETCH (UID {uid} RFC822.SIZE {reported_size} BODY[] {{{}}}\r\n", + raw.len() + ) + .into_bytes(); + response.extend_from_slice(raw); + response.extend_from_slice(b")\r\n{TAG} OK exact fetch completed\r\n"); + response + } + + async fn connect_fake_transport(server: &MockImapServerHandle) -> SessionUidOnlyTransport { + let account = AccountModel { + email: "synthetic-user".into(), + imap: Some(ImapConfig { + host: server.host(), + port: server.port(), + encryption: Encryption::None, + auth: AuthConfig { + auth_type: AuthType::Password, + password: None, + }, + use_proxy: None, + }), + ..Default::default() + }; + let limits = protocol_limits(64 * 1024).expect("bounded protocol limits"); + let client = Client::connection( + &server.host(), + &Encryption::None, + server.port(), + None, + false, + ) + .await + .expect("localhost connection"); + let (client, handle) = client + .with_uidonly(limits.clone()) + .expect("install UIDONLY guard"); + let mut session = client + .login("synthetic-user", "synthetic-secret") + .await + .expect("synthetic login"); + let capabilities = session.capabilities().await.expect("capabilities"); + assert!(capabilities.has_str("UIDONLY")); + assert!(capabilities.has_str("PARTIAL")); + assert!(capabilities.has_str("MESSAGELIMIT=2")); + session + .run_command_and_check_ok("ENABLE UIDONLY") + .await + .expect("enable UIDONLY"); + handle.ensure_active().expect("UIDONLY confirmed"); + SessionUidOnlyTransport { + session, + handle, + connection_scope: connection_scope(&account).unwrap(), + account, + limits, + } + } + + #[tokio::test] + async fn tcp_fake_yahoo_uidonly_pages_sparse_uids_and_checkpoints_after_verification() { + let messages = [ + (2, b"".as_slice()), + (7, b"Subject: seven\r\n\r\nseven".as_slice()), + (9, b"Subject: nine\r\n\r\nnine".as_slice()), + ]; + let server = MockImapServer::new() + .greeting("* OK synthetic Yahoo-like IMAP ready\r\n") + .respond("LOGIN", "{TAG} OK LOGIN completed\r\n") + .respond( + "CAPABILITY", + "* CAPABILITY IMAP4rev1 ENABLE UIDONLY PARTIAL MESSAGELIMIT=2\r\n{TAG} OK CAPABILITY completed\r\n", + ) + .respond( + "ENABLE UIDONLY", + "* ENABLED UIDONLY\r\n{TAG} OK ENABLE completed\r\n", + ) + .respond("EXAMINE", examine_response("Synthetic", 3, 77, 10)) + .respond("UID FETCH 1:9", inventory_response(&[(2, 999), (7, 1)])) + .respond("UID FETCH 8:9", inventory_response(&[(9, 0)])) + // The reported sizes are deliberately advisory and wrong. The + // literal byte count is the acquisition/accounting authority. + .respond("UID FETCH 2 (", exact_response(2, 1, messages[0].1)) + .respond("UID FETCH 7 (", exact_response(7, 2, messages[1].1)) + .respond("UID FETCH 9 (", exact_response(9, 3, messages[2].1)) + .start() + .await; + let mut transport = connect_fake_transport(&server).await; + let mut archive = FakeArchive::default(); + let report = run_acquisition( + &mut transport, + &mut archive, + "Synthetic", + Some(77), + limits(), + CancellationToken::new(), + |_| Ok(()), + ) + .await + .expect("complete fixed snapshot"); + + assert_eq!(report.checkpoint, Some(9)); + assert_eq!((report.inventoried, report.archived), (3, 3)); + assert_eq!(archive.projected, [2, 7, 9]); + + let commands = server.commands(); + let commands = commands.join("\n"); + assert_eq!(commands.matches(" EXAMINE ").count(), 2); + assert!(commands.contains(" UID FETCH 1:9 (UID RFC822.SIZE) (PARTIAL 1:2)")); + assert!(commands.contains(" UID FETCH 8:9 (UID RFC822.SIZE) (PARTIAL 1:2)")); + assert_eq!(commands.matches("BODY.PEEK[]").count(), 3); + let commands = commands.to_ascii_uppercase(); + assert!( + ![" STORE ", " MOVE ", " COPY ", " DELETE ", " EXPUNGE", " CLOSE",] + .iter() + .any(|forbidden| commands.contains(forbidden)) + ); + } +} diff --git a/crates/core/src/import/mod.rs b/crates/core/src/import/mod.rs index bd8db2ae..917c399b 100644 --- a/crates/core/src/import/mod.rs +++ b/crates/core/src/import/mod.rs @@ -122,6 +122,7 @@ impl ImportEmls { uid_next: None, uid_validity: None, highest_uid: None, + uidonly_source_scope: None, }; let mailbox_id = mailbox.id; // Upsert the mailbox, creating it if it doesn't exist @@ -413,6 +414,7 @@ pub(super) fn resolve_mailbox(account: &AccountModel, folder: &str) -> BichonRes uid_next: None, uid_validity: None, highest_uid: None, + uidonly_source_scope: None, }; let mailbox_id = mailbox.id; MailBox::batch_upsert(&[mailbox])?; diff --git a/crates/smtp/src/server.rs b/crates/smtp/src/server.rs index ad38ff48..4f3b06b8 100644 --- a/crates/smtp/src/server.rs +++ b/crates/smtp/src/server.rs @@ -656,6 +656,7 @@ async fn parse_email(data: &[u8], session: &Session) -> BichonResult<()> { uid_next: None, uid_validity: None, highest_uid: None, + uidonly_source_scope: None, }; if let Err(e) = MailBox::batch_upsert(&[mailbox]) { diff --git a/web/src/api/account/api.ts b/web/src/api/account/api.ts index 7be68c51..1bd48c2e 100644 --- a/web/src/api/account/api.ts +++ b/web/src/api/account/api.ts @@ -153,6 +153,7 @@ export interface AccountModel { created_at: number; updated_at: number; use_dangerous: boolean; + uidonly_enabled: boolean; pgp_key?: string; imap_quota_window?: QuotaWindow; imap_quota_bytes?: number; @@ -224,4 +225,4 @@ export const autoconfig = async (email: string) => { export const access_assign = async (data: Record) => { const response = await axiosInstance.post("api/v1/accounts/access/assignments", data); return response.data; -}; \ No newline at end of file +}; diff --git a/web/src/features/accounts/account-new.tsx b/web/src/features/accounts/account-new.tsx index 70e6cdc6..b90e394f 100644 --- a/web/src/features/accounts/account-new.tsx +++ b/web/src/features/accounts/account-new.tsx @@ -51,6 +51,7 @@ const defaultValues: AccountFormValues = { }, enabled: true, use_dangerous: false, + uidonly_enabled: false, date_since: undefined, date_before: undefined, download_interval_min: 60, @@ -126,6 +127,7 @@ export function AccountNewPage() { }, enabled: data.enabled, use_dangerous: data.use_dangerous, + uidonly_enabled: data.uidonly_enabled, date_since: data.date_since, date_before: data.date_before, download_interval_min: data.download_interval_min, diff --git a/web/src/features/accounts/account-settings-page.tsx b/web/src/features/accounts/account-settings-page.tsx index 1bcb8896..556935e7 100644 --- a/web/src/features/accounts/account-settings-page.tsx +++ b/web/src/features/accounts/account-settings-page.tsx @@ -57,6 +57,7 @@ function mapAccountToFormValues(account: AccountModel): AccountFormValues { imap, enabled: account.enabled, use_dangerous: account.use_dangerous, + uidonly_enabled: account.uidonly_enabled, date_since: account.date_since ?? undefined, date_before: account.date_before ?? undefined, download_interval_min: account.download_interval_min ?? 60, @@ -150,6 +151,7 @@ export function AccountSettingsPage({ accountId }: AccountSettingsPageProps) { }, enabled: data.enabled, use_dangerous: data.use_dangerous, + uidonly_enabled: data.uidonly_enabled, date_since: data.date_since, date_before: data.date_before, download_interval_min: data.download_interval_min, diff --git a/web/src/features/accounts/components/__tests__/schema-edge.test.ts b/web/src/features/accounts/components/__tests__/schema-edge.test.ts index c4568640..d1f186e9 100644 --- a/web/src/features/accounts/components/__tests__/schema-edge.test.ts +++ b/web/src/features/accounts/components/__tests__/schema-edge.test.ts @@ -16,6 +16,7 @@ const baseData = { }, enabled: true, use_dangerous: false, + uidonly_enabled: false, download_interval_min: 60, download_batch_size: 30, auto_download_new_mailboxes: true, diff --git a/web/src/features/accounts/components/__tests__/schema.test.ts b/web/src/features/accounts/components/__tests__/schema.test.ts index 8110f976..9195a137 100644 --- a/web/src/features/accounts/components/__tests__/schema.test.ts +++ b/web/src/features/accounts/components/__tests__/schema.test.ts @@ -16,6 +16,7 @@ const validAccountData = { }, enabled: true, use_dangerous: false, + uidonly_enabled: false, download_interval_min: 60, download_batch_size: 30, auto_download_new_mailboxes: true, diff --git a/web/src/features/accounts/components/schema.ts b/web/src/features/accounts/components/schema.ts index 3baa3200..8ad7ed9f 100644 --- a/web/src/features/accounts/components/schema.ts +++ b/web/src/features/accounts/components/schema.ts @@ -90,6 +90,7 @@ export const getAccountSchema = (isEdit: boolean, t: (key: string) => string) => imap: getImapConfigSchema(isEdit, t), enabled: z.boolean(), use_dangerous: z.boolean(), + uidonly_enabled: z.boolean(), date_since: dateSelectionSchema(t).optional(), date_before: relativeDateSchema(t).optional(), download_interval_min: z diff --git a/web/src/features/accounts/components/tab-server.tsx b/web/src/features/accounts/components/tab-server.tsx index 9e9c91a0..931231bd 100644 --- a/web/src/features/accounts/components/tab-server.tsx +++ b/web/src/features/accounts/components/tab-server.tsx @@ -209,6 +209,22 @@ export function TabServer({ isEdit }: TabServerProps) { )} /> + + ( + + + + +

+ {t('accounts.uidonlyEnabled')} + {t('accounts.uidonlyEnabledDescription')} +
+ + )} + /> ); } diff --git a/web/src/locales/en.json b/web/src/locales/en.json index 176e792a..8e6b1dce 100644 --- a/web/src/locales/en.json +++ b/web/src/locales/en.json @@ -376,6 +376,8 @@ "updateFailed": "Update failed, please try again later", "updateTheEmailAccountHere": "Update the email account here. ", "updatedAt": "Updated At", + "uidonlyEnabled": "Enable UID-only acquisition", + "uidonlyEnabledDescription": "Required for large Yahoo mailboxes. This changes remote message identity for this account; disabling it later stops synchronization instead of falling back to the limited mailbox view.", "useDangerous": "Trust Any TLS Certificate", "useDangerousDescription": "Enable this option only if you are connecting to an IMAP server with a public or self-signed certificate that may not be recognized by your system. Using this setting bypasses standard certificate validation, which can expose you to man-in-the-middle attacks. Only enable if you understand the risks.", "useNoProxy": "No Proxy", @@ -1825,4 +1827,4 @@ "singleRequestBatchSizeTooLarge": "Batch size must be at most 200", "singleRequestBatchSizeTooSmall": "Batch size must be at least 10" } -} \ No newline at end of file +}