External/bichon
1
0
Fork 0
mirror of https://github.com/rustmailer/bichon.git synced 2026-08-03 07:48:34 +02:00
Code Issues Packages Projects Releases Wiki Activity

perf: optimize IMAP account fetch flow

Browse Source
This commit is contained in:
rustmailer
2026-05-26 15:23:34 +08:00
parent f30cd66e00
commit 83dd9cdd6b
7 changed files with 221 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/core/src/cache/imap/download/flow.rs
Vendored
+103 -105
View File
@@ -54,7 +54,7 @@ pub async fn fetch_and_save_by_date(
mailbox: &MailBox,
direction: FetchDirection,
token: CancellationToken,
) -> BichonResult<()> {
) -> BichonResult<Option<u32>> {
let account_id = account.id;
let mut session = match ImapExecutor::create_connection(account_id).await {
Ok(session) => session,
@@ -108,32 +108,18 @@ pub async fn fetch_and_save_by_date(
FolderStatus::Success,
None,
)?;
return Ok(());
return Ok(None);
}
let folder_limit = account.folder_limit;
// sort small -> bigger
let mut uid_vec: Vec<u32> = uid_list.into_iter().collect();
uid_vec.sort();
if let Some(limit) = folder_limit {
let limit = limit.max(100) as usize;
if len > limit {
uid_vec = match direction {
FetchDirection::Since => uid_vec.split_off(len - limit),
FetchDirection::Before => {
uid_vec.truncate(limit);
uid_vec
}
};
}
}
let max_uid = uid_vec.last().copied();
let planned = uid_vec.len() as u64;
let uid_batches = generate_uid_sequence_hashset(
uid_vec,
account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize,
false,
);
DownloadState::update_folder_progress(
account_id,
@@ -212,14 +198,16 @@ pub async fn fetch_and_save_by_date(
)?;
}
session.logout().await.ok();
Ok(())
Ok(max_uid)
}
/// Fetches all messages from a mailbox.
/// Returns `Ok(Some(max_uid))` with the highest UID stored, or `Ok(None)` if empty.
pub async fn fetch_and_save_full_mailbox(
account: &AccountModel,
mailbox: &MailBox,
token: CancellationToken,
) -> BichonResult<()> {
) -> BichonResult<Option<u32>> {
let mailbox_id = mailbox.id;
let account_id = account.id;
@@ -262,33 +250,17 @@ pub async fn fetch_and_save_full_mailbox(
}
};
let folder_limit = account.folder_limit;
let total_to_fetch = match folder_limit {
Some(limit) if (limit as u64) < total => {
let limit64 = limit as u64;
total.min(limit64.max(100))
}
_ => total,
};
let page_size = if let Some(limit) = folder_limit {
limit
.max(100)
.min(account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE))
} else {
account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE)
};
let total_batches = total_to_fetch.div_ceil(page_size as u64);
let desc = folder_limit.is_some();
let page_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE);
let total_batches = total.div_ceil(page_size as u64);
info!(
"Starting full mailbox download for '{}', total={}, limit={:?}, batches={}, desc={}",
mailbox.name, total, folder_limit, total_batches, desc
"Starting full mailbox download for '{}', total={}, batches={}",
mailbox.name, total, total_batches
);
let mut current_processed = 0u64;
let mut has_error_or_cancel = false;
let mut max_uid: Option<u32> = None;
for page in 1..=total_batches {
if token.is_cancelled() {
@@ -300,7 +272,7 @@ pub async fn fetch_and_save_full_mailbox(
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total_to_fetch,
total,
current_processed,
FolderStatus::Cancelled,
None,
@@ -313,12 +285,12 @@ pub async fn fetch_and_save_full_mailbox(
&mut session,
account_id,
mailbox_id,
total_to_fetch,
total,
page as u64,
page_size as u64,
&mailbox.encoded_name(),
desc,
token.clone(),
&mut max_uid,
)
.await
{
@@ -327,7 +299,7 @@ pub async fn fetch_and_save_full_mailbox(
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total_to_fetch,
total,
current_processed,
FolderStatus::Downloading,
None,
@@ -339,7 +311,7 @@ pub async fn fetch_and_save_full_mailbox(
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total_to_fetch,
total,
current_processed,
FolderStatus::Failed,
Some(err_msg),
@@ -354,28 +326,24 @@ pub async fn fetch_and_save_full_mailbox(
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total_to_fetch,
total,
current_processed,
FolderStatus::Success,
None,
)?;
}
session.logout().await.ok();
Ok(())
Ok(max_uid)
}
pub fn generate_uid_sequence_hashset(
unique_nums: Vec<u32>,
chunk_size: usize,
desc: bool,
) -> Vec<(String, u64)> {
assert!(!unique_nums.is_empty());
let mut nums = unique_nums;
if desc {
nums.reverse();
}
let mut result = Vec::new();
let nums = unique_nums;
for chunk in nums.chunks(chunk_size) {
let size = chunk.len() as u64;
@@ -448,7 +416,7 @@ pub async fn reconcile_mailboxes(
break;
}
if local_mailbox.uid_validity != remote_mailbox.uid_validity {
let new_highest_uid = if local_mailbox.uid_validity != remote_mailbox.uid_validity {
if remote_mailbox.uid_validity.is_none() {
let err_msg = format!(
"Mailbox '{}' logic error: Server did not provide UIDVALIDITY.",
@@ -493,7 +461,7 @@ pub async fn reconcile_mailboxes(
FetchDirection::Since,
token.clone(),
)
.await?;
.await?
}
None => match &account.date_before {
Some(r) => {
@@ -505,7 +473,7 @@ pub async fn reconcile_mailboxes(
FetchDirection::Before,
token.clone(),
)
.await?;
.await?
}
None => {
rebuild_mailbox_cache(
@@ -520,10 +488,12 @@ pub async fn reconcile_mailboxes(
}
} else {
perform_incremental_sync(account, local_mailbox, remote_mailbox, token.clone())
.await?;
}
.await?
};
mailboxes_to_update.push(remote_mailbox.clone());
let mut updated = remote_mailbox.clone();
updated.highest_uid = new_highest_uid;
mailboxes_to_update.push(updated);
}
//The metadata of this mailbox must only be updated after a successful synchronization;
//otherwise, it may cause synchronization errors and result in missing emails in the local sync results.
@@ -598,7 +568,11 @@ pub async fn reconcile_mailboxes(
};
match result {
Ok(_) => {}
Ok(new_highest_uid) => {
let mut updated = mailbox.clone();
updated.highest_uid = new_highest_uid;
MailBox::batch_upsert(&[updated])?;
}
Err(err) => {
has_error = true;
tracing::error!("Folder sync task failed: {:#?}", err);
@@ -622,64 +596,88 @@ pub async fn reconcile_mailboxes(
}
//only check new emails and sync
/// Incrementally syncs a mailbox.
/// Returns the new highest UID after sync, or `None` if nothing changed.
async fn perform_incremental_sync(
account: &AccountModel,
local_mailbox: &MailBox,
remote_mailbox: &MailBox,
token: CancellationToken,
) -> BichonResult<()> {
) -> BichonResult<Option<u32>> {
if remote_mailbox.exists > 0 {
let local_max_uid = ENVELOPE_MANAGER.get_max_uid(account.id, local_mailbox.id)?;
tracing::info!(
"[account {}][mailbox {}] perform_incremental_sync: local_max_uid={:?}, remote.exists={}",
account.id,
local_mailbox.name,
local_max_uid,
remote_mailbox.exists
);
match local_max_uid {
Some(max_uid) => {
let mut session = ImapExecutor::create_connection(account.id).await?;
let before_date = account
.date_before
.as_ref()
.map(|r| r.calculate_date())
.transpose()?;
ImapExecutor::fetch_new_mail(
&mut session,
account,
local_mailbox,
max_uid + 1,
before_date.as_deref(),
token,
)
.await?;
session.logout().await.ok();
// 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
}
None => {
info!(
"No maximum UID found in index for mailbox, assuming local cache is missing."
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
);
match &account.date_since {
Some(date_since) => {
fetch_and_save_by_date(
account,
date_since.since_date()?.as_str(),
remote_mailbox,
FetchDirection::Since,
token,
)
.await?;
}
match local_max_uid {
Some(uid) => uid + 1,
None => {
fetch_and_save_full_mailbox(account, remote_mailbox, token).await?;
info!(
"No maximum UID found in index for mailbox, assuming local cache is missing."
);
let result = match &account.date_since {
Some(date_since) => {
fetch_and_save_by_date(
account,
date_since.since_date()?.as_str(),
remote_mailbox,
FetchDirection::Since,
token,
)
.await?
}
None => {
fetch_and_save_full_mailbox(account, remote_mailbox, token)
.await?
}
};
return Ok(result);
}
}
}
}
}
};
Ok(())
let mut session = ImapExecutor::create_connection(account.id).await?;
let before_date = account
.date_before
.as_ref()
.map(|r| r.calculate_date())
.transpose()?;
let new_max_uid = ImapExecutor::fetch_new_mail(
&mut session,
account,
local_mailbox,
start_uid,
before_date.as_deref(),
token,
)
.await?;
session.logout().await.ok();
// Keep existing highest_uid if no new mail was fetched.
Ok(new_max_uid.or(local_mailbox.highest_uid))
} else {
Ok(local_mailbox.highest_uid)
}
}
crates/core/src/cache/imap/download/rebuild.rs
Vendored
+18 -10
View File
@@ -89,7 +89,11 @@ pub async fn rebuild_cache(
};
match fetch_and_save_full_mailbox(&account, &mailbox, token.clone()).await {
Ok(_) => {}
Ok(new_highest_uid) => {
let mut updated = mailbox.clone();
updated.highest_uid = new_highest_uid;
MailBox::batch_upsert(&[updated])?;
}
Err(err) => {
has_error = true;
tracing::error!("Folder sync task failed: {:#?}", err);
@@ -169,7 +173,11 @@ pub async fn rebuild_cache_by_date(
match fetch_and_save_by_date(&account, date.as_str(), &mailbox, direction, token.clone())
.await
{
Ok(_) => {}
Ok(new_highest_uid) => {
let mut updated = mailbox.clone();
updated.highest_uid = new_highest_uid;
MailBox::batch_upsert(&[updated])?;
}
Err(err) => {
has_error = true;
tracing::error!("Folder sync task failed: {:#?}", err);
@@ -196,7 +204,7 @@ pub async fn rebuild_mailbox_cache(
local_mailbox: &MailBox,
remote_mailbox: &MailBox,
token: CancellationToken,
) -> BichonResult<()> {
) -> BichonResult<Option<u32>> {
ENVELOPE_MANAGER
.delete_mailbox_envelopes(account.id, vec![local_mailbox.id])
.await?;
@@ -217,11 +225,11 @@ pub async fn rebuild_mailbox_cache(
FolderStatus::Success,
None,
)?;
return Ok(());
return Ok(None);
}
fetch_and_save_full_mailbox(account, remote_mailbox, token).await?;
Ok(())
let result = fetch_and_save_full_mailbox(account, remote_mailbox, token).await?;
Ok(result)
}
pub async fn rebuild_mailbox_cache_by_date(
@@ -231,7 +239,7 @@ pub async fn rebuild_mailbox_cache_by_date(
remote: &MailBox,
direction: FetchDirection,
token: CancellationToken,
) -> BichonResult<()> {
) -> BichonResult<Option<u32>> {
ENVELOPE_MANAGER
.delete_mailbox_envelopes(account.id, vec![local_mailbox_id])
.await?;
@@ -252,9 +260,9 @@ pub async fn rebuild_mailbox_cache_by_date(
FolderStatus::Success,
None,
)?;
return Ok(());
return Ok(None);
}
fetch_and_save_by_date(account, date, remote, direction, token).await?;
Ok(())
let result = fetch_and_save_by_date(account, date, remote, direction, token).await?;
Ok(result)
}
crates/core/src/cache/imap/mailbox.rs
Vendored
+4
View File
@@ -56,6 +56,10 @@ pub struct MailBox {
/// The validity identifier for UIDs in this mailbox, used to ensure UID consistency across sessions.
/// If `None`, the IMAP server has not provided this information.
pub uid_validity: Option<u32>,
/// The highest UID that has been successfully downloaded and stored locally.
/// 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<u32>,
}
impl MemDbModel for MailBox {
crates/core/src/imap/executor.rs
+82 -117
View File
@@ -17,8 +17,7 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use crate::account::migration::AccountModel;
use crate::account::state::{DownloadState, FolderStatus};
use crate::cache::imap::download::flow::{generate_uid_sequence_hashset, DEFAULT_BATCH_SIZE};
use crate::account::state::{DownloadState, DownloadStatus, FolderStatus};
use crate::cache::imap::mailbox::MailBox;
use crate::envelope::extractor::extract_envelope_and_store_it;
use crate::error::code::ErrorCode;
@@ -80,6 +79,14 @@ impl ImapExecutor {
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))
}
/// Fetches new mail for a mailbox by directly issuing a ranged UID FETCH.
///
/// Unlike the old two-step approach (UID SEARCH → batch UID FETCH),
/// this sends a single `UID FETCH {start}:*` and streams the results,
/// eliminating one IMAP round-trip.
///
/// Returns `Ok(Some(max_uid))` with the highest UID fetched, or `Ok(None)`
/// if no new mail was found.
pub async fn fetch_new_mail(
session: &mut Session<Box<dyn SessionStream>>,
account: &AccountModel,
@@ -87,122 +94,90 @@ impl ImapExecutor {
start_uid: u64,
before: Option<&str>,
token: CancellationToken,
) -> BichonResult<()> {
) -> BichonResult<Option<u32>> {
assert!(start_uid > 0, "start_uid must be greater than 0");
let query = match before {
Some(date) => format!("UID {start_uid}:* BEFORE {date}"),
None => format!("UID {start_uid}:*"),
// Select the mailbox (read-only) so UID FETCH works.
session
.examine(&mailbox.encoded_name())
.await
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
// Build the UID range. IMAP UID FETCH accepts the same range syntax
// as UID SEARCH, so we can skip the separate SEARCH round-trip.
let uid_range = match before {
Some(date) => format!("{start_uid}:* BEFORE {date}"),
None => format!("{start_uid}:*"),
};
let uid_list = match Self::uid_search(session, &mailbox.encoded_name(), &query).await {
Ok(uid_list) => uid_list,
Err(e) => {
let err_msg = format!("UID search failed in [{}]: {:#?}", mailbox.name, e);
DownloadState::update_folder_progress(
info!(
"[account {}][mailbox {}] fetch_new_mail: direct 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), ErrorCode::ImapCommandFailed)
})?;
let mut count = 0u64;
let mut max_uid: Option<u32> = None;
while let Some(fetch) = stream
.try_next()
.await
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
{
if token.is_cancelled() {
tracing::info!(
"Account {}: fetch_new_mail stream interrupted.",
account.id
);
DownloadState::update_session_status(
account.id,
mailbox.name.clone(),
0,
0,
FolderStatus::Failed,
Some(err_msg.clone()),
DownloadStatus::Cancelled,
Some("User stopped or system shutdown".to_string()),
)?;
DownloadState::append_session_error(account.id, err_msg)?;
return Err(e);
return Err(raise_error!(
"Stream cancelled".into(),
ErrorCode::InternalError
));
}
};
let len = uid_list.len();
if len == 0 {
let msg = match before {
Some(date) => format!("No emails found before {}.", date),
None => "No new emails found.".into(),
};
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;
}
if count == 0 {
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
0,
0,
FolderStatus::Success,
Some(msg),
Some("No new emails found.".into()),
)?;
return Ok(());
}
info!(
"[account {}][mailbox {}] {} envelopes need to be fetched (start_uid={})",
account.id, mailbox.name, len, start_uid
);
tracing::debug!(
"[account {}][mailbox {}] fetch_new_mail UID range: {}..{} ({} uids)",
account.id,
mailbox.name,
start_uid,
uid_list.iter().max().unwrap_or(&0),
len
);
let mut uid_vec: Vec<u32> = uid_list.into_iter().collect();
uid_vec.sort();
let uid_batches = generate_uid_sequence_hashset(
uid_vec,
account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize,
false,
);
let mut current_processed = 0u64;
let mut has_error_or_cancel = false;
for (index, batch) in uid_batches.into_iter().enumerate() {
if token.is_cancelled() {
break;
}
match Self::uid_batch_retrieve_emails(
session,
account.id,
mailbox.id,
&batch.0,
token.clone(),
)
.await
{
Ok(_) => {
current_processed += batch.1;
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
len as u64,
current_processed,
FolderStatus::Downloading,
None,
)?;
}
Err(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(),
len as u64,
current_processed,
FolderStatus::Failed,
Some(err_msg),
)?;
has_error_or_cancel = true;
break;
}
}
}
if !has_error_or_cancel {
} else {
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
len as u64,
current_processed,
count,
count,
FolderStatus::Success,
None,
)?;
}
Ok(())
Ok(max_uid)
}
pub async fn batch_retrieve_emails(
@@ -213,36 +188,23 @@ impl ImapExecutor {
page: u64,
page_size: u64,
encoded_mailbox_name: &str,
desc: bool,
token: CancellationToken,
max_uid: &mut Option<u32>,
) -> BichonResult<usize> {
assert!(page > 0, "Page number must be greater than 0");
assert!(page_size > 0, "Page size must be greater than 0");
let (start, end) = if desc {
// Fetch messages starting from the newest (descending order)
let end = total.saturating_sub((page - 1) * page_size);
if end == 0 {
return Ok(0);
}
// Calculate start as end - page_size + 1 to avoid off-by-one errors
let start = end.saturating_sub(page_size - 1).max(1);
(start, end)
} else {
// Fetch messages starting from the oldest (ascending order)
let start = (page - 1) * page_size + 1;
if start > total {
return Ok(0);
}
// Calculate end, capped by the total number of messages
let end = (start + page_size - 1).min(total);
(start, end)
};
// 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 {}, desc={})",
encoded_mailbox_name, sequence_set, page, page_size, desc
"Fetching mailbox '{}' messages: sequence {} (page {}, page_size {})",
encoded_mailbox_name, sequence_set, page, page_size
);
let mut stream = session
@@ -263,6 +225,9 @@ impl ImapExecutor {
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;
}
crates/core/src/import/mod.rs
+1
View File
@@ -105,6 +105,7 @@ impl ImportEmls {
unseen: None,
uid_next: None,
uid_validity: None,
highest_uid: None,
};
let mailbox_id = mailbox.id;
// Upsert the mailbox, creating it if it doesn't exist
crates/core/src/mailbox/list.rs
+12 -2
View File
@@ -157,8 +157,13 @@ async fn fetch_remote_with_progress(account_id: u64) -> BichonResult<Vec<MailBox
mailbox.account_id = account_id;
mailbox.id = create_hash(account_id, &mailbox.name);
// Use STATUS instead of EXAMINE: gets MESSAGES/UNSEEN/UIDNEXT/UIDVALIDITY
// without selecting the mailbox, avoiding context switches.
let mx = session
.examine(mailbox_name.as_str())
.status(
mailbox_name.as_str(),
"(MESSAGES UNSEEN UIDNEXT UIDVALIDITY)",
)
.await
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
mailbox.exists = mx.exists;
@@ -205,8 +210,13 @@ pub async fn convert_names_to_mailboxes(
mailbox.account_id = account_id;
mailbox.id = create_hash(account_id, &mailbox.name);
// Use STATUS instead of EXAMINE: gets MESSAGES/UNSEEN/UIDNEXT/UIDVALIDITY
// without selecting the mailbox, avoiding context switches.
let mx = session
.examine(mailbox_name.as_str())
.status(
mailbox_name.as_str(),
"(MESSAGES UNSEEN UIDNEXT UIDVALIDITY)",
)
.await
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
mailbox.exists = mx.exists;
crates/smtp/src/server.rs
+1
View File
@@ -628,6 +628,7 @@ async fn parse_email(data: &[u8], session: &Session) -> BichonResult<()> {
unseen: None,
uid_next: None,
uid_validity: None,
highest_uid: None,
};
let mailbox_id = mailbox.id;