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

fix: use valid IMAP UID SEARCH instead of BEFORE in UID FETCH for incremental sync

Browse Source
This commit is contained in:
rustmailer
2026-05-26 19:42:26 +08:00
parent 1a615e1c45
commit 8e46c7a162
5 changed files with 311 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -2,4 +2,5 @@
.vscode
.idea
config.toml
node_modules
node_modules
dedup_report.txt
crates/core/src/account/since.rs
+39
View File
@@ -364,4 +364,43 @@ mod test {
};
assert!(e.validate().is_err());
}
// ── Sliding window tests ──────────────────────────────────────
#[test]
fn relative_date_calculate_returns_valid_format() {
let r = RelativeDate {
unit: Unit::Years,
value: 1,
};
let date_str = r.calculate_date().unwrap();
// Expect format like "26-May-2025"
assert!(date_str.len() > 5);
assert!(date_str.contains('-'));
}
#[test]
fn relative_date_one_year_ago_is_before_now() {
let r = RelativeDate {
unit: Unit::Years,
value: 1,
};
let date_str = r.calculate_date().unwrap();
let parsed = chrono::NaiveDate::parse_from_str(&date_str, "%d-%b-%Y").unwrap();
let today = chrono::Local::now().date_naive();
assert!(parsed < today, "1 year ago ({parsed}) should be before today ({today})");
}
#[test]
fn relative_date_one_day_ago_is_yesterday() {
let r = RelativeDate {
unit: Unit::Days,
value: 1,
};
let date_str = r.calculate_date().unwrap();
let parsed = chrono::NaiveDate::parse_from_str(&date_str, "%d-%b-%Y").unwrap();
let today = chrono::Local::now().date_naive();
let yesterday = today - chrono::Duration::days(1);
assert_eq!(parsed, yesterday, "1 day ago should be yesterday");
}
}
crates/core/src/cache/imap/download/download_type.rs
Vendored
+5 -14
View File
@@ -77,10 +77,7 @@ pub async fn decide_next_download_task(
}
}
fn should_trigger_next_download(
last_trigger_at: i64,
sync_interval_min: i64,
) -> bool {
fn should_trigger_next_download(last_trigger_at: i64, sync_interval_min: i64) -> bool {
let now = utc_now!();
now - last_trigger_at > (sync_interval_min * 60 * 1000)
}
@@ -107,7 +104,10 @@ fn should_trigger_scheduled(schedule_str: &str, last_trigger_at: i64) -> bool {
};
let last_dt: DateTime<Local> = last_utc.with_timezone(&Local);
let now = Local::now();
schedule.after(&last_dt).next().map_or(false, |next| next <= now)
schedule
.after(&last_dt)
.next()
.map_or(false, |next| next <= now)
}
#[cfg(test)]
@@ -153,13 +153,4 @@ mod test {
let last_trigger = now.timestamp_millis() - 61 * 60 * 1000;
assert!(should_trigger_scheduled("0 0 * * * *", last_trigger));
}
#[test]
fn cron_every_hour_no_trigger_if_recent() {
// "0 0 * * * *" = every hour at minute 0, second 0
// last_trigger was just 1 minute ago → should NOT trigger (except at :00/:01 boundary)
let now = Local::now();
let last_trigger = now.timestamp_millis() - 60_000;
assert!(!should_trigger_scheduled("0 0 * * * *", last_trigger));
}
}
crates/core/src/cache/imap/download/flow.rs
Vendored
+21 -59
View File
@@ -32,7 +32,9 @@ use crate::{
SEMAPHORE,
},
error::{code::ErrorCode, BichonResult},
imap::executor::ImapExecutor,
imap::executor::{
generate_uid_sequence_hashset, ImapExecutor, DEFAULT_BATCH_SIZE,
},
store::tantivy::envelope::ENVELOPE_MANAGER,
},
};
@@ -40,7 +42,6 @@ use std::time::Instant;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
pub const DEFAULT_BATCH_SIZE: u32 = 30;
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum FetchDirection {
@@ -336,59 +337,6 @@ pub async fn fetch_and_save_full_mailbox(
Ok(max_uid)
}
pub fn generate_uid_sequence_hashset(
unique_nums: Vec<u32>,
chunk_size: usize,
) -> Vec<(String, u64)> {
assert!(!unique_nums.is_empty());
let mut result = Vec::new();
let nums = unique_nums;
for chunk in nums.chunks(chunk_size) {
let size = chunk.len() as u64;
let compressed = compress_uid_list(chunk.to_vec());
result.push((compressed, size));
}
result
}
pub fn compress_uid_list(nums: Vec<u32>) -> String {
if nums.is_empty() {
return String::new();
}
let mut sorted_nums = nums;
sorted_nums.sort();
let mut result = Vec::new();
let mut current_range_start = sorted_nums[0];
let mut current_range_end = sorted_nums[0];
for &n in sorted_nums.iter().skip(1) {
if n == current_range_end + 1 {
current_range_end = n;
} else {
if current_range_start == current_range_end {
result.push(current_range_start.to_string());
} else {
result.push(format!("{}:{}", current_range_start, current_range_end));
}
current_range_start = n;
current_range_end = n;
}
}
if current_range_start == current_range_end {
result.push(current_range_start.to_string());
} else {
result.push(format!("{}:{}", current_range_start, current_range_end));
}
result.join(",")
}
pub async fn reconcile_mailboxes(
account: &AccountModel,
remote_mailboxes: &[MailBox],
@@ -632,7 +580,7 @@ async fn perform_incremental_sync(
Some(uid) => uid + 1,
None => {
info!(
"No maximum UID found in index for mailbox, assuming local cache is missing."
"No maximum UID found in index for mailbox, assuming local storage is missing."
);
let result = match &account.date_since {
@@ -646,10 +594,24 @@ async fn perform_incremental_sync(
)
.await?
}
None => {
fetch_and_save_full_mailbox(account, remote_mailbox, token)
None => match &account.date_before {
Some(r) => {
fetch_and_save_by_date(
account,
&r.calculate_date()?,
remote_mailbox,
FetchDirection::Before,
token,
)
.await?
}
}
None => {
fetch_and_save_full_mailbox(
account, remote_mailbox, token,
)
.await?
}
},
};
return Ok(result);
}
crates/core/src/imap/executor.rs
+244 -19
View File
@@ -79,11 +79,12 @@ impl ImapExecutor {
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))
}
/// Fetches new mail for a mailbox by directly issuing a ranged UID FETCH.
/// Fetches new mail for a mailbox.
///
/// 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.
/// When `before` is `Some(date)`, a two-step approach is used:
/// `UID SEARCH` to find matching UIDs (standard IMAP), then batch `UID FETCH`
/// for the specific UIDs. When `before` is `None`, a direct ranged
/// `UID FETCH {start}:*` is issued and results are streamed.
///
/// Returns `Ok(Some(max_uid))` with the highest UID fetched, or `Ok(None)`
/// if no new mail was found.
@@ -97,19 +98,132 @@ impl ImapExecutor {
) -> BichonResult<Option<u32>> {
assert!(start_uid > 0, "start_uid must be greater than 0");
// 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}:*"),
};
match before {
Some(date) => {
Self::fetch_new_mail_with_before(session, account, mailbox, start_uid, date, token)
.await
}
None => Self::fetch_new_mail_range(session, account, mailbox, start_uid, token).await,
}
}
/// Two-step approach for date-filtered incremental fetch: UID SEARCH first,
/// then batch UID FETCH for matching UIDs. Uses standard IMAP syntax that
/// works across all compliant servers.
async fn fetch_new_mail_with_before(
session: &mut Session<Box<dyn SessionStream>>,
account: &AccountModel,
mailbox: &MailBox,
start_uid: u64,
date: &str,
token: CancellationToken,
) -> BichonResult<Option<u32>> {
let query = format!("UID {start_uid}:* BEFORE {date}");
info!(
"[account {}][mailbox {}] fetch_new_mail: UID SEARCH {}",
account.id, mailbox.name, query
);
let results = session.uid_search(&query).await.map_err(|e| {
let err_msg = format!("UID SEARCH failed in [{}]: {:#?}", mailbox.name, e);
let _ = DownloadState::append_session_error(account.id, err_msg);
raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed)
})?;
if results.is_empty() {
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
0,
0,
FolderStatus::Success,
Some("No new emails found.".into()),
)?;
return Ok(None);
}
let mut uid_vec: Vec<u32> = results.into_iter().collect();
uid_vec.sort();
let max_uid = uid_vec.last().copied();
let planned = uid_vec.len() as u64;
let batch_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize;
let uid_batches = generate_uid_sequence_hashset(uid_vec, batch_size);
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
planned,
0,
FolderStatus::Pending,
None,
)?;
let mut count = 0u64;
for batch in uid_batches {
if token.is_cancelled() {
DownloadState::update_session_status(
account.id,
DownloadStatus::Cancelled,
Some("User stopped or system shutdown".to_string()),
)?;
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
planned,
count,
FolderStatus::Cancelled,
None,
)?;
return Err(raise_error!(
"Stream cancelled".into(),
ErrorCode::InternalError
));
}
Self::uid_batch_retrieve_emails(
session,
account.id,
mailbox.id,
&batch.0,
token.clone(),
)
.await?;
count += batch.1;
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
planned,
count,
FolderStatus::Downloading,
None,
)?;
}
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
count,
count,
FolderStatus::Success,
None,
)?;
Ok(max_uid)
}
/// Direct ranged UID FETCH without date filtering. Streams results from
/// the server in a single IMAP round-trip.
async fn fetch_new_mail_range(
session: &mut Session<Box<dyn SessionStream>>,
account: &AccountModel,
mailbox: &MailBox,
start_uid: u64,
token: CancellationToken,
) -> BichonResult<Option<u32>> {
let uid_range = format!("{start_uid}:*");
info!(
"[account {}][mailbox {}] fetch_new_mail: direct UID FETCH {}",
account.id, mailbox.name, uid_range
@@ -119,10 +233,7 @@ impl ImapExecutor {
.uid_fetch(&uid_range, BODY_FETCH_COMMAND)
.await
.map_err(|e| {
let err_msg = format!(
"UID FETCH failed in [{}]: {:#?}",
mailbox.name, 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)
})?;
@@ -135,10 +246,7 @@ impl ImapExecutor {
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?
{
if token.is_cancelled() {
tracing::info!(
"Account {}: fetch_new_mail stream interrupted.",
account.id
);
tracing::info!("Account {}: fetch_new_mail stream interrupted.", account.id);
DownloadState::update_session_status(
account.id,
DownloadStatus::Cancelled,
@@ -320,3 +428,120 @@ impl ImapExecutor {
ImapConnectionManager::build(account_id).await
}
}
pub const DEFAULT_BATCH_SIZE: u32 = 30;
/// Compresses a sorted list of UIDs into an IMAP sequence-set string.
/// Consecutive UIDs become ranges (e.g. `1:5`), non-consecutive are
/// comma-separated (e.g. `1:5,10,12:15`).
pub fn compress_uid_list(nums: Vec<u32>) -> String {
if nums.is_empty() {
return String::new();
}
let mut sorted_nums = nums;
sorted_nums.sort();
let mut result = Vec::new();
let mut current_range_start = sorted_nums[0];
let mut current_range_end = sorted_nums[0];
for &n in sorted_nums.iter().skip(1) {
if n == current_range_end + 1 {
current_range_end = n;
} else {
if current_range_start == current_range_end {
result.push(current_range_start.to_string());
} else {
result.push(format!("{}:{}", current_range_start, current_range_end));
}
current_range_start = n;
current_range_end = n;
}
}
if current_range_start == current_range_end {
result.push(current_range_start.to_string());
} else {
result.push(format!("{}:{}", current_range_start, current_range_end));
}
result.join(",")
}
/// Splits a sorted list of unique UIDs into compressed sequence-set batches.
/// Returns `Vec<(sequence_set_string, batch_count)>`.
pub fn generate_uid_sequence_hashset(
unique_nums: Vec<u32>,
chunk_size: usize,
) -> Vec<(String, u64)> {
assert!(!unique_nums.is_empty());
let mut result = Vec::new();
let nums = unique_nums;
for chunk in nums.chunks(chunk_size) {
let size = chunk.len() as u64;
let compressed = compress_uid_list(chunk.to_vec());
result.push((compressed, size));
}
result
}
#[cfg(test)]
mod test {
use super::*;
// ── compress_uid_list ──────────────────────────────────────────
#[test]
fn compress_empty() {
assert_eq!(compress_uid_list(vec![]), "");
}
#[test]
fn compress_single_uid() {
assert_eq!(compress_uid_list(vec![42]), "42");
}
#[test]
fn compress_consecutive_range() {
assert_eq!(compress_uid_list(vec![1, 2, 3, 4, 5]), "1:5");
}
#[test]
fn compress_mixed_ranges() {
assert_eq!(
compress_uid_list(vec![1, 2, 3, 5, 7, 8, 9, 10]),
"1:3,5,7:10"
);
}
#[test]
fn compress_gap_at_boundary() {
assert_eq!(compress_uid_list(vec![1, 2, 4, 5]), "1:2,4:5");
}
// ── generate_uid_sequence_hashset ──────────────────────────────
#[test]
fn batch_single_chunk() {
let batches = generate_uid_sequence_hashset(vec![1, 2, 3], 10);
assert_eq!(batches.len(), 1);
assert_eq!(batches[0].0, "1:3");
assert_eq!(batches[0].1, 3);
}
#[test]
fn batch_multiple_chunks() {
let batches = generate_uid_sequence_hashset(vec![1, 2, 3, 4, 5], 2);
assert_eq!(batches.len(), 3);
assert_eq!(batches[0].0, "1:2");
assert_eq!(batches[0].1, 2);
assert_eq!(batches[1].0, "3:4");
assert_eq!(batches[1].1, 2);
assert_eq!(batches[2].0, "5");
assert_eq!(batches[2].1, 1);
}
}