majjo/bichon
1
0
Fork 0
mirror of https://github.com/rustmailer/bichon.git synced 2026-08-31 01:52:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(imap): resilient batched sync, stale-session cleanup, and live progress UI

Browse Source 
- Add SyncFull trigger and configurable IMAP socket read timeout
  - Replace streaming UID FETCH with UID SEARCH ALL + batched fetch so per-message progress stays responsive and throttling servers can retry with reconnection
  - Finalize stale Running sessions on startup so interrupted syncs no longer show a phantom "syncing" state
  - Show a live syncing pill on the account row; add elapsed time, current folder, and slow-server warning styling in the dialog
This commit is contained in:
rustmailer
2026-08-04 17:47:42 +08:00
parent 9b1f6cae8c
commit 75d345b742
30 changed files with 831 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
crates/core/src/account/state.rs
View File

@@ -40,6 +40,10 @@ pub enum TriggerType {
Manual,
#[default]
Scheduled,
/// Full re-sync (UID SEARCH ALL) invoked explicitly, e.g. to repair a
/// mailbox whose incremental download was interrupted. Semantically a
/// manual trigger, tracked distinctly for diagnostics.
SyncFull,
}
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
@@ -189,6 +193,47 @@ impl DownloadState {
})
}
/// Moves a stale Running session into history as Cancelled.
///
/// A Running `active_session` that survives an Idle decision means the
/// previous run was interrupted without a clean shutdown (e.g. process
/// killed mid-download). Leaving it in place makes the UI show a phantom
/// "syncing" state even though nothing is downloading. Callers invoke this
/// only when no download is actually running for the account, so a
/// legitimately active session is never touched.
///
/// Returns `true` if a stale session was finalized (i.e. the previous sync
/// did not finish) and `false` otherwise.
pub fn finalize_stale_session(account_id: u64) -> BichonResult<bool> {
let stale = Self::get(account_id)?
.and_then(|s| s.active_session)
.map_or(false, |s| s.status == DownloadStatus::Running);
if stale {
Self::update_state(account_id, move |current| {
let mut updated = current.clone();
if updated
.active_session
.as_ref()
.map_or(false, |s| s.status == DownloadStatus::Running)
{
if let Some(mut session) = updated.active_session.take() {
session.status = DownloadStatus::Cancelled;
session.end_time = Some(utc_now!());
session.message = Some(
"Previous sync did not finish cleanly; marked as cancelled on startup."
.into(),
);
updated.history.push(session);
updated.last_finished_at = Some(utc_now!());
updated.active_session = None;
}
}
Ok(updated)
})?;
}
Ok(stale)
}
pub fn update_folder_progress(
account_id: u64,
folder_name: String,
@@ -220,6 +265,19 @@ impl DownloadState {
})
}
/// Touches only `current_folder` without rewriting folder progress. Lets
/// long-running IMAP operations (e.g. waiting on a slow server mid-batch)
/// keep the UI's "last activity" indicator fresh without spamming writes.
pub fn set_current_folder(account_id: u64, folder_name: String) -> BichonResult<()> {
Self::update_state(account_id, move |state| {
let mut updated = state.clone();
if let Some(ref mut session) = updated.active_session {
session.current_folder = Some(folder_name);
}
Ok(updated)
})
}
pub fn init_folder_details(account_id: u64, folders: Vec<String>) -> BichonResult<()> {
Self::update_state(account_id, move |state| {
let mut updated = state.clone();

7
crates/core/src/cache/imap/download/download_type.rs vendored
View File

@@ -53,6 +53,7 @@ pub async fn decide_next_download_task(
let should_start = match trigger_type {
TriggerType::Manual => true,
TriggerType::SyncFull => true,
TriggerType::Scheduled => {
let now = utc_now!();
let cooldown_ok = now - state.last_finished_at.unwrap_or(0) > 60 * 1000;
@@ -73,10 +74,14 @@ pub async fn decide_next_download_task(
DownloadState::start_new_session(account.id, trigger_type)?;
Ok(DownloadTask::TraceFetch)
} else {
// Nothing to download right now. A Running active_session at this point
// is a leftover from an interrupted run (a real download would have
// been blocked by the busy guard before reaching here), so mark it
// Cancelled instead of leaving the UI showing a phantom "syncing".
DownloadState::finalize_stale_session(account.id)?;
Ok(DownloadTask::Idle)
}
}
fn should_trigger_next_download(last_trigger_at: i64, sync_interval_min: i64) -> bool {
let now = utc_now!();
now - last_trigger_at > (sync_interval_min * 60 * 1000)

403
crates/core/src/cache/imap/download/flow.rs vendored
View File

@@ -33,7 +33,8 @@ use crate::{
},
error::{code::ErrorCode, BichonResult},
imap::executor::{
compress_uid_list, generate_uid_sequence_hashset, ImapExecutor, DEFAULT_BATCH_SIZE,
compress_uid_list, generate_uid_sequence_hashset, slow_server_message, ImapExecutor,
DEFAULT_BATCH_SIZE,
},
store::tantivy::envelope::ENVELOPE_MANAGER,
},
@@ -44,7 +45,6 @@ use tracing::{debug, error, info, warn};
const MAX_NETWORK_RETRIES: u32 = 3;
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum FetchDirection {
Since,
@@ -163,13 +163,24 @@ pub async fn fetch_and_save_by_date(
&batch.0,
account.max_email_size_bytes,
token.clone(),
Some(&|cumulative, avg_secs, stall_secs| {
// Per-message progress: the current batch's cumulative count
// keeps the UI moving while a slow server trickles messages.
let _ = DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
planned,
cumulative,
FolderStatus::Downloading,
slow_server_message(avg_secs, stall_secs),
);
Ok(())
}),
)
.await
{
Ok(processed) => break Ok(processed),
Err(e)
if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError =>
{
Ok((processed, _throttled)) => break Ok(processed),
Err(e) if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError => {
retries += 1;
warn!(
account_id,
@@ -183,22 +194,16 @@ pub async fn fetch_and_save_by_date(
match ImapExecutor::create_connection(account_id).await {
Ok(new_session) => {
session = new_session;
if let Err(e2) = session.examine(&mailbox.encoded_name()).await
{
let err_msg = format!(
"Re-examine failed after reconnect: {:#?}",
e2
);
DownloadState::append_session_error(
account_id,
err_msg,
)?;
if let Err(e2) = session.examine(&mailbox.encoded_name()).await {
let err_msg =
format!("Re-examine failed after reconnect: {:#?}", e2);
DownloadState::append_session_error(account_id, err_msg)?;
break Err(e);
}
tokio::time::sleep(Duration::from_secs(
1 << (retries - 1),
))
.await;
// Longer backoff than the original 1s/2s/4s: a
// throttling server needs time to recover.
let backoff = [5u64, 15, 30][(retries - 1) as usize];
tokio::time::sleep(Duration::from_secs(backoff)).await;
continue;
}
Err(e2) => {
@@ -254,6 +259,11 @@ pub async fn fetch_and_save_by_date(
/// Fetches all messages from a mailbox.
/// Returns `Ok(Some(max_uid))` with the highest UID stored, or `Ok(None)` if empty.
///
/// The mailbox is enumerated via `UID SEARCH ALL` first, then downloaded in UID
/// batches. Unlike sequence-number paging, UIDs stay stable while the download
/// runs (new arrivals only get larger UIDs), so no message is silently skipped
/// when the server changes mid-download.
pub async fn fetch_and_save_full_mailbox(
account: &AccountModel,
mailbox: &MailBox,
@@ -279,41 +289,57 @@ pub async fn fetch_and_save_full_mailbox(
}
};
let total = match session.examine(&mailbox.encoded_name()).await {
Ok(mailbox) => mailbox.exists as u64,
Err(e) => {
let err_msg = format!("Failed to examine folder [{}]: {:#?}", mailbox.name, e);
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
mailbox.exists as u64,
0,
FolderStatus::Failed,
Some(err_msg.clone()),
)?;
let uid_list =
match ImapExecutor::uid_search_all_mailbox(&mut session, &mailbox.encoded_name()).await {
Ok(list) => list,
Err(e) => {
let err_msg = format!("UID SEARCH failed in [{}]: {:#?}", mailbox.name, e);
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
0,
0,
FolderStatus::Failed,
Some(err_msg.clone()),
)?;
DownloadState::append_session_error(account_id, err_msg)?;
session.logout().await.ok();
return Err(e);
}
};
DownloadState::append_session_error(account_id, err_msg)?;
session.logout().await.ok();
return Err(raise_error!(
format!("{:#?}", e),
ErrorCode::ImapCommandFailed
));
}
};
let planned = uid_list.len() as u64;
if planned == 0 {
info!(
"Mailbox '{}' is empty, no emails to download.",
mailbox.name
);
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
0,
0,
FolderStatus::Success,
None,
)?;
session.logout().await.ok();
return Ok(None);
}
let page_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE);
let total_batches = total.div_ceil(page_size as u64);
let max_uid = *uid_list.last().unwrap();
let page_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize;
let uid_batches = generate_uid_sequence_hashset(uid_list, page_size);
let total_batches = uid_batches.len();
info!(
"Starting full mailbox download for '{}', total={}, batches={}",
mailbox.name, total, total_batches
"Starting full mailbox download for '{}', uids={}, batches={}",
mailbox.name, planned, total_batches
);
let mut current_processed = 0u64;
let mut has_error_or_cancel = false;
let mut max_uid: Option<u32> = None;
for page in 1..=total_batches {
for (index, batch) in uid_batches.into_iter().enumerate() {
if token.is_cancelled() {
DownloadState::update_session_status(
account_id,
@@ -323,7 +349,7 @@ pub async fn fetch_and_save_full_mailbox(
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total,
planned,
current_processed,
FolderStatus::Cancelled,
None,
@@ -332,48 +358,66 @@ pub async fn fetch_and_save_full_mailbox(
break;
}
// Heartbeat: keeps `current_folder` fresh so the web UI can show
// "waiting for server" while a slow IMAP server is mid-batch.
DownloadState::set_current_folder(account_id, mailbox.name.clone())?;
// Heartbeat: keeps `current_folder` fresh so the web UI can show
// "waiting for server" while a slow IMAP server is mid-batch.
DownloadState::set_current_folder(account_id, mailbox.name.clone())?;
let mut retries = 0u32;
let batch_result = loop {
match ImapExecutor::batch_retrieve_emails(
match ImapExecutor::uid_batch_retrieve_emails(
&mut session,
account_id,
mailbox_id,
total,
page as u64,
page_size as u64,
&mailbox.encoded_name(),
&batch.0,
account.max_email_size_bytes,
token.clone(),
&mut max_uid,
Some(&|cumulative, avg_secs, stall_secs| {
// Per-message progress: the current batch's cumulative count
// keeps the UI moving while a slow server trickles messages.
let _ = DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
planned,
cumulative,
FolderStatus::Downloading,
slow_server_message(avg_secs, stall_secs),
);
Ok(())
}),
)
.await
{
Ok(count) => break Ok(count),
Err(e)
if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError =>
{
Ok((count, _throttled)) => break Ok(count),
Err(e) if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError => {
retries += 1;
warn!(
account_id,
mailbox = mailbox.name,
page,
index,
retries,
"Network error on batch, reconnecting ({}/{})",
retries,
MAX_NETWORK_RETRIES
);
// Refresh the heartbeat after a reconnect too.
let _ = DownloadState::set_current_folder(account_id, mailbox.name.clone());
match ImapExecutor::create_connection(account_id).await {
Ok(new_session) => {
session = new_session;
if let Err(e2) = session.examine(&mailbox.encoded_name()).await {
let err_msg = format!(
"Re-examine failed after reconnect: {:#?}",
e2
);
let err_msg =
format!("Re-examine failed after reconnect: {:#?}", e2);
DownloadState::append_session_error(account_id, err_msg)?;
break Err(e);
}
tokio::time::sleep(Duration::from_secs(1 << (retries - 1))).await;
// Longer backoff than the original 1s/2s/4s: a
// throttling server needs time to recover.
let backoff = [5u64, 15, 30][(retries - 1) as usize];
tokio::time::sleep(Duration::from_secs(backoff)).await;
continue;
}
Err(e2) => {
@@ -391,19 +435,19 @@ pub async fn fetch_and_save_full_mailbox(
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total,
planned,
current_processed,
FolderStatus::Downloading,
None,
)?;
}
Err(e) => {
let err_msg = format!("Batch {} failed: {:#?}", page, e);
let err_msg = format!("Batch {} failed: {:#?}", index, e);
DownloadState::append_session_error(account_id, err_msg.clone())?;
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total,
planned,
current_processed,
FolderStatus::Failed,
Some(err_msg),
@@ -411,21 +455,21 @@ pub async fn fetch_and_save_full_mailbox(
has_error_or_cancel = true;
break;
}
};
}
}
if !has_error_or_cancel {
DownloadState::update_folder_progress(
account_id,
mailbox.name.clone(),
total,
planned,
current_processed,
FolderStatus::Success,
None,
)?;
}
session.logout().await.ok();
Ok(max_uid)
Ok(max_uid.into())
}
/// Generates a synthetic UIDVALIDITY for IMAP servers that don't provide it.
@@ -490,15 +534,13 @@ where
Ok(None) => {
warn!(
attempt = attempt + 1,
max_retries,
"STATUS returned no UIDVALIDITY"
max_retries, "STATUS returned no UIDVALIDITY"
);
}
Err(e) => {
warn!(
attempt = attempt + 1,
max_retries,
"UIDVALIDITY fetch attempt failed: {:#?}", e
max_retries, "UIDVALIDITY fetch attempt failed: {:#?}", e
);
}
}
@@ -634,9 +676,7 @@ async fn reconcile_uid_validity_change(
.await
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let batch_size = account
.download_batch_size
.unwrap_or(DEFAULT_BATCH_SIZE) as usize;
let batch_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize;
let batches = generate_uid_sequence_hashset(missing_uids, batch_size);
let mut downloaded = 0u64;
@@ -661,10 +701,23 @@ async fn reconcile_uid_validity_change(
&batch.0,
account.max_email_size_bytes,
token.clone(),
Some(&|cumulative, avg_secs, stall_secs| {
// Per-message progress: the current batch's cumulative count
// keeps the UI moving while a slow server trickles messages.
let _ = DownloadState::update_folder_progress(
account_id,
remote_mailbox.name.clone(),
planned,
cumulative,
FolderStatus::Downloading,
slow_server_message(avg_secs, stall_secs),
);
Ok(())
}),
)
.await
{
Ok(processed) => {
Ok((processed, _throttled)) => {
downloaded += processed;
DownloadState::update_folder_progress(
account_id,
@@ -791,18 +844,12 @@ pub async fn reconcile_mailboxes(
account_id, local_mailbox.name, &local_mailbox.uid_validity, &remote_uid_validity
);
reconcile_uid_validity_change(
account,
local_mailbox,
remote_mailbox,
token.clone(),
)
.await?
reconcile_uid_validity_change(account, local_mailbox, remote_mailbox, token.clone())
.await?
} else {
perform_incremental_sync(account, local_mailbox, remote_mailbox, token.clone())
.await?
};
let mut updated = remote_mailbox.clone();
updated.highest_uid = new_highest_uid;
// Update uid_validity with the resolved value (either from server or synthetic)
@@ -924,33 +971,15 @@ async fn perform_incremental_sync(
// Use stored highest_uid if available; otherwise fall back to Tantivy
// query once (backward compatibility with pre-existing databases).
let start_uid = match local_mailbox.highest_uid {
Some(uid) => {
tracing::info!(
"[account {}][mailbox {}] perform_incremental_sync: stored highest_uid={}, remote.exists={}",
account.id,
local_mailbox.name,
uid,
remote_mailbox.exists
);
uid as u64 + 1
}
Some(uid) => uid as u64 + 1,
None => {
let local_max_uid =
ENVELOPE_MANAGER.get_max_uid(account.id, local_mailbox.id)?;
tracing::info!(
"[account {}][mailbox {}] perform_incremental_sync: highest_uid unset, Tantivy max_uid={:?}, remote.exists={}",
account.id,
local_mailbox.name,
local_max_uid,
remote_mailbox.exists
);
let local_max_uid = ENVELOPE_MANAGER.get_max_uid(account.id, local_mailbox.id)?;
match local_max_uid {
Some(uid) => uid + 1,
None => {
info!(
"No maximum UID found in index for mailbox, assuming local storage is missing."
);
let result = match &account.date_since {
Some(date_since) => {
fetch_and_save_by_date(
@@ -974,10 +1003,8 @@ async fn perform_incremental_sync(
.await?
}
None => {
fetch_and_save_full_mailbox(
account, remote_mailbox, token,
)
.await?
fetch_and_save_full_mailbox(account, remote_mailbox, token)
.await?
}
},
};
@@ -993,6 +1020,19 @@ async fn perform_incremental_sync(
.as_ref()
.map(|r| r.calculate_date())
.transpose()?;
if start_uid == 1 {
// No stored highest_uid and no indexed messages: fall back to a
// full mailbox download. Tell the UI up-front so it doesn't show
// a stale "Pending" while the (possibly large) mailbox streams.
let _ = DownloadState::update_folder_progress(
account.id,
remote_mailbox.name.clone(),
0,
0,
FolderStatus::Downloading,
Some("Full mailbox download".into()),
)?;
}
let new_max_uid = ImapExecutor::fetch_new_mail(
&mut session,
@@ -1034,7 +1074,10 @@ mod tests {
fn test_generate_synthetic_uidvalidity_different_mailboxes() {
let inbox = generate_synthetic_uidvalidity("INBOX");
let sent = generate_synthetic_uidvalidity("Sent");
assert_ne!(inbox, sent, "different mailboxes should have different uid_validity");
assert_ne!(
inbox, sent,
"different mailboxes should have different uid_validity"
);
}
#[test]
@@ -1070,10 +1113,8 @@ mod tests {
// Ensure a rustls crypto provider is installed (ring).
// May already be installed by production code; ignore duplicate.
rustls::crypto::CryptoProvider::install_default(
rustls::crypto::ring::default_provider(),
)
.ok();
rustls::crypto::CryptoProvider::install_default(rustls::crypto::ring::default_provider())
.ok();
let tcp = TcpStream::connect((host, port))
.await
@@ -1084,8 +1125,8 @@ mod tests {
let timeout_stream = TimeoutStream::new(tcp);
let pinned = Box::pin(timeout_stream);
let server_name = ServerName::try_from(host.to_owned())
.map_err(|e| format!("Invalid hostname: {e}"))?;
let server_name =
ServerName::try_from(host.to_owned()).map_err(|e| format!("Invalid hostname: {e}"))?;
let config = ClientConfig::builder()
.with_root_certificates(rustls::RootCertStore {
@@ -1208,11 +1249,7 @@ mod tests {
session.logout().await.ok();
println!(
"Call {}: UIDVALIDITY = {:?}",
i + 1,
status.uid_validity
);
println!("Call {}: UIDVALIDITY = {:?}", i + 1, status.uid_validity);
results.borrow_mut().push(status.uid_validity);
}
@@ -1275,13 +1312,10 @@ mod tests {
let sample: Vec<u32> = all_uids.into_iter().take(5).collect();
let uid_set = compress_uid_list(sample.clone());
let result = ImapExecutor::fetch_uid_metadata(
&mut session,
&uid_set,
CancellationToken::new(),
)
.await
.expect("fetch_uid_metadata should succeed");
let result =
ImapExecutor::fetch_uid_metadata(&mut session, &uid_set, CancellationToken::new())
.await
.expect("fetch_uid_metadata should succeed");
session.logout().await.ok();
@@ -1329,8 +1363,7 @@ mod tests {
#[tokio::test]
async fn test_retry_first_attempt_succeeds() {
let result = fetch_uid_validity_with_retry_inner(3, mock_results(vec![Ok(Some(42))]))
.await;
let result = fetch_uid_validity_with_retry_inner(3, mock_results(vec![Ok(Some(42))])).await;
assert_eq!(result.unwrap(), Some(42));
}
@@ -1392,13 +1425,7 @@ mod tests {
// max_retries=5, success on 5th attempt
let result = fetch_uid_validity_with_retry_inner(
5,
mock_results(vec![
Ok(None),
Ok(None),
Ok(None),
Ok(None),
Ok(Some(5)),
]),
mock_results(vec![Ok(None), Ok(None), Ok(None), Ok(None), Ok(Some(5))]),
)
.await;
assert_eq!(result.unwrap(), Some(5));
@@ -1418,11 +1445,7 @@ mod tests {
#[tokio::test]
async fn test_retry_max_retries_zero() {
// max_retries=0 means no attempts at all
let result = fetch_uid_validity_with_retry_inner(
0,
mock_results(vec![Ok(Some(42))]),
)
.await;
let result = fetch_uid_validity_with_retry_inner(0, mock_results(vec![Ok(Some(42))])).await;
assert_eq!(result.unwrap(), None);
}
@@ -1431,8 +1454,8 @@ mod tests {
// ============================================================
use crate::imap::mock_server::{
examine_response, uid_fetch_metadata_response, uid_fetch_rfc822_response,
minimal_eml, MockImapServer, MockImapServerHandle,
examine_response, minimal_eml, uid_fetch_metadata_response, uid_fetch_rfc822_response,
MockImapServer, MockImapServerHandle,
};
/// Build an `async_imap::Session` connected to the mock server,
@@ -1453,9 +1476,11 @@ mod tests {
client.read_response().await.unwrap();
// Login
let mut session = client.login("user", "pass").await.map_err(|(e, _)| {
panic!("Login failed: {e:?}")
}).unwrap();
let mut session = client
.login("user", "pass")
.await
.map_err(|(e, _)| panic!("Login failed: {e:?}"))
.unwrap();
// Examine
session.examine("INBOX").await.unwrap();
@@ -1468,37 +1493,28 @@ mod tests {
let handle = MockImapServer::new()
.respond("LOGIN", "{TAG} OK LOGIN done\r\n")
.respond("EXAMINE", examine_response("INBOX", 3, 42, 4))
.respond("UID FETCH", uid_fetch_metadata_response(&[
(1, "<msg-a@test.com>"),
(2, "<msg-b@test.com>"),
(3, "<msg-c@test.com>"),
]))
.respond(
"UID FETCH",
uid_fetch_metadata_response(&[
(1, "<msg-a@test.com>"),
(2, "<msg-b@test.com>"),
(3, "<msg-c@test.com>"),
]),
)
.start()
.await;
let mut session = mock_session(&handle).await;
let result = ImapExecutor::fetch_uid_metadata(
&mut session,
"1:3",
CancellationToken::new(),
)
.await
.unwrap();
let result =
ImapExecutor::fetch_uid_metadata(&mut session, "1:3", CancellationToken::new())
.await
.unwrap();
assert_eq!(result.len(), 3);
assert_eq!(
result.get(&1).unwrap().as_deref(),
Some("msg-a@test.com")
);
assert_eq!(
result.get(&2).unwrap().as_deref(),
Some("msg-b@test.com")
);
assert_eq!(
result.get(&3).unwrap().as_deref(),
Some("msg-c@test.com")
);
assert_eq!(result.get(&1).unwrap().as_deref(), Some("msg-a@test.com"));
assert_eq!(result.get(&2).unwrap().as_deref(), Some("msg-b@test.com"));
assert_eq!(result.get(&3).unwrap().as_deref(), Some("msg-c@test.com"));
session.logout().await.ok();
}
@@ -1523,10 +1539,7 @@ mod tests {
let mut session = mock_session(&handle).await;
let uids: Vec<u32> = {
let mut stream = session
.uid_fetch("1:2", "(UID FLAGS)")
.await
.unwrap();
let mut stream = session.uid_fetch("1:2", "(UID FLAGS)").await.unwrap();
use futures::TryStreamExt;
let mut uids = Vec::new();
@@ -1557,10 +1570,7 @@ mod tests {
let mut session = mock_session(&handle).await;
let bodies: Vec<(u32, Vec<u8>)> = {
let mut stream = session
.uid_fetch("1:1", "(UID BODY[])")
.await
.unwrap();
let mut stream = session.uid_fetch("1:1", "(UID BODY[])").await.unwrap();
use futures::TryStreamExt;
let mut bodies = Vec::new();
@@ -1591,18 +1601,12 @@ mod tests {
let mut session = mock_session(&handle).await;
let result = ImapExecutor::fetch_uid_metadata(
&mut session,
"1:*",
CancellationToken::new(),
)
.await
.unwrap();
let result =
ImapExecutor::fetch_uid_metadata(&mut session, "1:*", CancellationToken::new())
.await
.unwrap();
assert!(
result.is_empty(),
"empty mailbox should return empty map"
);
assert!(result.is_empty(), "empty mailbox should return empty map");
session.logout().await.ok();
}
@@ -1610,8 +1614,7 @@ mod tests {
#[tokio::test]
async fn fetch_uid_metadata_missing_message_id() {
// One entry has a Message-ID, the other has no header at all.
let header_with_msgid =
"From: sender@example.com\r\n\
let header_with_msgid = "From: sender@example.com\r\n\
Date: Thu, 01 Jan 2025 00:00:00 +0000\r\n\
Message-ID: <ok@test.com>\r\n\r\n";
let header_without_msgid = "\r\n";
@@ -1637,19 +1640,13 @@ mod tests {
let mut session = mock_session(&handle).await;
let result = ImapExecutor::fetch_uid_metadata(
&mut session,
"1:2",
CancellationToken::new(),
)
.await
.unwrap();
let result =
ImapExecutor::fetch_uid_metadata(&mut session, "1:2", CancellationToken::new())
.await
.unwrap();
assert_eq!(result.len(), 2);
assert_eq!(
result.get(&1).unwrap().as_deref(),
Some("ok@test.com")
);
assert_eq!(result.get(&1).unwrap().as_deref(), Some("ok@test.com"));
// UID 2 has no Message-ID header → None
assert_eq!(result.get(&2).unwrap().as_deref(), None);

25
crates/core/src/context/executors.rs
View File

@@ -17,6 +17,7 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use crate::account::migration::AccountType;
use crate::account::state::DownloadState;
use crate::context::Initialize;
use crate::{
{
@@ -25,7 +26,7 @@ use crate::{
utc_now,
};
use std::sync::LazyLock;
use tracing::info;
use tracing::{info, warn};
pub static BICHON_CONTEXT: LazyLock<BichonContext> = LazyLock::new(BichonContext::new);
@@ -65,6 +66,28 @@ impl BichonContext {
active_accounts.len()
);
for account in active_accounts {
// A Running session surviving startup is a leftover from a previous
// interrupted run; nothing is downloading yet at this point. Mark it
// Cancelled so the UI doesn't show a phantom "syncing" state. The
// scheduler starts regardless — its first tick runs immediately, so
// the interrupted run is caught up on, and the session's trigger
// stays Scheduled rather than showing a "Manual" the user never
// initiated.
match DownloadState::finalize_stale_session(account.id) {
Ok(true) => {
info!(
"Account {}: stale sync session finalized on startup.",
account.id
);
}
Err(e) => {
warn!(
"Failed to finalize stale session for account {}: {:#?}",
account.id, e
);
}
Ok(false) => {}
}
DOWNLOAD_CONTROLLER
.trigger_schedule(account.id, account.email)
.await

482
crates/core/src/imap/executor.rs
View File

@@ -33,6 +33,7 @@ use tracing::info;
const BODY_FETCH_COMMAND: &str = "(UID INTERNALDATE RFC822.SIZE BODY.PEEK[])";
const SIZE_ONLY_FETCH: &str = "(UID RFC822.SIZE)";
const MAX_NETWORK_RETRIES: u32 = 3;
fn classify_imap_error(e: &async_imap::error::Error) -> ErrorCode {
match e {
@@ -97,6 +98,30 @@ impl ImapExecutor {
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))
}
/// Enumerate every UID currently present in the mailbox via `UID SEARCH ALL`.
///
/// This is the drift-safe way to page through a full mailbox: UIDs are stable
/// while the download runs (new arrivals only get larger UIDs), whereas
/// sequence numbers shift when mail is added or removed mid-download, which
/// silently skips messages. The caller owns the returned list and decides
/// how to batch it (see `generate_uid_sequence_hashset`).
pub async fn uid_search_all_mailbox(
session: &mut Session<Box<dyn SessionStream>>,
mailbox_name: &str,
) -> BichonResult<Vec<u32>> {
session
.examine(mailbox_name)
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?;
let results = session
.uid_search("ALL")
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?;
let mut uids: Vec<u32> = results.into_iter().collect();
uids.sort();
Ok(uids)
}
/// Fetches new mail for a mailbox.
///
/// When `before` is `Some(date)`, a two-step approach is used:
@@ -201,15 +226,33 @@ impl ImapExecutor {
ErrorCode::InternalError
));
}
let processed = Self::uid_batch_retrieve_emails(
let (processed, throttled) = Self::uid_batch_retrieve_emails(
session,
account.id,
mailbox.id,
&batch.0,
account.max_email_size_bytes,
token.clone(),
Some(&|cumulative, avg_secs, stall_secs| {
// Per-message progress: the current batch's cumulative count
// keeps the UI moving while a slow server trickles messages.
let _ = DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
planned,
cumulative,
FolderStatus::Downloading,
slow_server_message(avg_secs, stall_secs),
);
Ok(())
}),
)
.await?;
if throttled {
// Server appears to be rate-limiting; back off before the next
// batch so we don't hammer the limiter with back-to-back bursts.
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
}
count += processed;
DownloadState::update_folder_progress(
account.id,
@@ -233,8 +276,15 @@ impl ImapExecutor {
Ok(max_uid)
}
/// Direct ranged UID FETCH without date filtering. Streams results from
/// the server in a single IMAP round-trip.
/// Fetches all messages with UID >= start_uid via batched UID FETCH.
///
/// A single ranged `UID FETCH {start}:*` can block for minutes on slow
/// servers pushing hundreds of messages, and hits the socket read timeout
/// if the server stalls, with zero progress feedback in the meantime.
/// Instead, enumerate the UIDs first, then download in small batches —
/// each batch is a short round-trip with a SIZE pre-check (oversized
/// messages are skipped without fetching their body), progress is reported
/// per batch, and the whole download stays responsive to cancellation.
async fn fetch_new_mail_range(
session: &mut Session<Box<dyn SessionStream>>,
account: &AccountModel,
@@ -244,32 +294,48 @@ impl ImapExecutor {
) -> BichonResult<Option<u32>> {
let uid_range = format!("{start_uid}:*");
info!(
"[account {}][mailbox {}] fetch_new_mail: direct UID FETCH {}",
"[account {}][mailbox {}] fetch_new_mail: batched UID FETCH {}",
account.id, mailbox.name, uid_range
);
let mut stream = session
.uid_fetch(&uid_range, BODY_FETCH_COMMAND)
.await
.map_err(|e| {
let err_msg = format!("UID FETCH failed in [{}]: {:#?}", mailbox.name, e);
let _ = DownloadState::append_session_error(account.id, err_msg);
raise_error!(format!("{:#?}", e), classify_imap_error(&e))
})?;
let results = session.uid_search(&uid_range).await.map_err(|e| {
let err_msg = format!("UID SEARCH failed in [{}]: {:#?}", mailbox.name, e);
let _ = DownloadState::append_session_error(account.id, err_msg);
raise_error!(format!("{:#?}", e), classify_imap_error(&e))
})?;
let mut uid_vec: Vec<u32> = results.into_iter().collect();
uid_vec.sort();
if uid_vec.is_empty() {
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
0,
0,
FolderStatus::Success,
Some("No new emails found.".into()),
)?;
return Ok(None);
}
let max_uid = uid_vec.last().copied();
let planned = uid_vec.len() as u64;
let batch_size = account.download_batch_size.unwrap_or(DEFAULT_BATCH_SIZE) as usize;
let batches = generate_uid_sequence_hashset(uid_vec, batch_size);
let total_batches = batches.len();
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
planned,
0,
FolderStatus::Downloading,
None,
)?;
let mut count = 0u64;
let mut skipped = 0u64;
let mut max_uid: Option<u32> = None;
let size_limit = account
.max_email_size_bytes
.unwrap_or(DEFAULT_MAX_EMAIL_SIZE);
while let Some(fetch) = stream
.try_next()
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?
{
for (index, batch) in batches.into_iter().enumerate() {
if token.is_cancelled() {
tracing::info!("Account {}: fetch_new_mail stream interrupted.", account.id);
DownloadState::update_session_status(
account.id,
DownloadStatus::Cancelled,
@@ -281,149 +347,134 @@ impl ImapExecutor {
));
}
let msg_size = fetch.size.unwrap_or(0) as u64;
if msg_size > 0 && msg_size > size_limit {
tracing::warn!(
account_id = account.id,
mailbox_id = mailbox.id,
uid = fetch.uid,
size = msg_size,
limit = size_limit,
"Skipping oversized email (streaming mode)"
);
skipped += 1;
continue;
// A slow server can stall a batch past the socket read timeout.
// Retry such batches on a fresh connection instead of failing the
// whole sync session.
let mut retries = 0u32;
let batch_result = loop {
match Self::uid_batch_retrieve_emails(
session,
account.id,
mailbox.id,
&batch.0,
account.max_email_size_bytes,
token.clone(),
Some(&|cumulative, avg_secs, stall_secs| {
// Report per-message so the UI moves even while a slow
// server trickles out the current batch.
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
planned,
cumulative,
FolderStatus::Downloading,
slow_server_message(avg_secs, stall_secs),
)
}),
)
.await
{
Ok(processed) => break Ok(processed),
Err(e)
if retries < MAX_NETWORK_RETRIES && e.code() == ErrorCode::NetworkError =>
{
retries += 1;
tracing::warn!(
account_id = account.id,
mailbox = mailbox.name,
index,
retries,
"Network error on batch, reconnecting ({}/{})",
retries,
MAX_NETWORK_RETRIES
);
match ImapExecutor::create_connection(account.id).await {
Ok(new_session) => {
*session = new_session;
if let Err(e2) = session.examine(&mailbox.encoded_name()).await {
let err_msg =
format!("Re-examine failed after reconnect: {:#?}", e2);
DownloadState::append_session_error(account.id, err_msg)?;
break Err(e);
}
// Longer backoff than the original 1s/2s/4s: a
// throttling server needs time to recover.
let backoff = [5u64, 15, 30][(retries - 1) as usize];
tracing::warn!(
account_id = account.id,
mailbox = mailbox.name,
"Backing off {}s before retrying batch",
backoff
);
tokio::time::sleep(std::time::Duration::from_secs(backoff)).await;
continue;
}
Err(e2) => {
tracing::error!(
account_id = account.id,
"Reconnection failed: {:#?}",
e2
);
break Err(e);
}
}
}
Err(e) => break Err(e),
}
};
match batch_result {
Ok((processed, _throttled)) => {
count += processed;
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
planned,
count,
FolderStatus::Downloading,
None,
)?;
tracing::debug!(
"[account {}][mailbox {}] fetch_new_mail: batch {}/{} done ({} processed)",
account.id,
mailbox.name,
index + 1,
total_batches,
processed
);
}
Err(e) => {
DownloadState::append_session_error(account.id, format!("{:#?}", e))?;
return Err(e);
}
}
if let Some(uid) = fetch.uid {
max_uid = Some(max_uid.unwrap_or(0).max(uid));
if count == planned {
break;
}
extract_envelope_and_store_it(fetch, account.id, mailbox.id).await?;
count += 1;
}
let total = count + skipped;
if total == 0 {
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
0,
0,
FolderStatus::Success,
Some("No new emails found.".into()),
)?;
} else {
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
total,
count,
FolderStatus::Success,
if skipped > 0 {
Some(format!("{skipped} email(s) skipped due to size limit"))
} else {
None
},
)?;
}
DownloadState::update_folder_progress(
account.id,
mailbox.name.clone(),
count,
count,
FolderStatus::Success,
None,
)?;
Ok(max_uid)
}
pub async fn batch_retrieve_emails(
session: &mut Session<Box<dyn SessionStream>>,
account_id: u64,
mailbox_id: u64,
total: u64,
page: u64,
page_size: u64,
encoded_mailbox_name: &str,
max_email_size_bytes: Option<u64>,
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");
// Fetch messages starting from the oldest (ascending order).
let start = (page - 1) * page_size + 1;
if start > total {
return Ok(0);
}
let end = (start + page_size - 1).min(total);
let sequence_set = format!("{}:{}", start, end);
info!(
"Fetching mailbox '{}' messages: sequence {} (page {}, page_size {})",
encoded_mailbox_name, sequence_set, page, page_size
);
let limit = max_email_size_bytes.unwrap_or(DEFAULT_MAX_EMAIL_SIZE);
// PASS 1: fetch only SIZE to identify oversized messages
let acceptable_uids = {
let mut size_stream = session
.fetch(sequence_set.as_str(), SIZE_ONLY_FETCH)
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?;
let mut uids: Vec<u32> = Vec::new();
while let Some(fetch) = size_stream
.try_next()
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?
{
let uid = fetch.uid.unwrap_or(0);
let msg_size = fetch.size.unwrap_or(0) as u64;
if msg_size == 0 || msg_size <= limit {
uids.push(uid);
} else {
tracing::warn!(
account_id,
mailbox_id,
uid,
size = msg_size,
limit,
"Skipping oversized email"
);
}
}
uids
};
if acceptable_uids.is_empty() {
return Ok(0);
}
// PASS 2: fetch bodies only for acceptable UIDs
let filtered = compress_uid_list(acceptable_uids);
let mut body_stream = session
.uid_fetch(&filtered, BODY_FETCH_COMMAND)
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?;
let mut count = 0;
while let Some(fetch) = body_stream
.try_next()
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?
{
if token.is_cancelled() {
tracing::info!("Account {}: UID fetch stream interrupted.", account_id);
return Err(raise_error!(
"Stream cancelled".into(),
ErrorCode::InternalError
));
}
if let Some(uid) = fetch.uid {
*max_uid = Some((*max_uid).unwrap_or(0).max(uid));
}
extract_envelope_and_store_it(fetch, account_id, mailbox_id).await?;
count += 1;
}
Ok(count)
}
/// Downloads the bodies of `uid_set` in one batch.
///
/// `progress` (if given) is invoked after each stored message with the
/// cumulative count for the whole mailbox, the current average
/// inter-message interval in seconds (None until at least two messages
/// arrived), and the current stall duration in seconds when the server is
/// silent (None when a message just arrived). The UI can then move per
/// message — and detect a slow server — instead of only updating when a
/// batch finishes. Slow servers push a batch's messages over many seconds
/// (even minutes); without per-message updates the UI freezes and looks
/// stuck.
pub async fn uid_batch_retrieve_emails(
session: &mut Session<Box<dyn SessionStream>>,
account_id: u64,
@@ -431,7 +482,10 @@ impl ImapExecutor {
uid_set: &str,
max_email_size_bytes: Option<u64>,
token: CancellationToken,
) -> BichonResult<u64> {
progress: Option<
&(dyn Fn(u64, Option<f64>, Option<f64>) -> BichonResult<()> + Send + Sync),
>,
) -> BichonResult<(u64, bool)> {
let limit = max_email_size_bytes.unwrap_or(DEFAULT_MAX_EMAIL_SIZE);
// PASS 1: fetch only SIZE to identify oversized messages
@@ -466,7 +520,7 @@ impl ImapExecutor {
};
if acceptable_uids.is_empty() {
return Ok(0);
return Ok((0, false));
}
// PASS 2: fetch bodies only for acceptable UIDs
@@ -477,11 +531,43 @@ impl ImapExecutor {
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?;
let mut count = 0u64;
while let Some(fetch) = body_stream
.try_next()
.await
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?
{
// Sliding window of recent per-message receive times, used to estimate
// how slow the server is: slow servers push messages seconds apart.
let mut recv_times: std::collections::VecDeque<std::time::Instant> =
std::collections::VecDeque::with_capacity(11);
let mut last_recv = std::time::Instant::now();
// Consecutive stall reports. A high value means the server is
// throttling us; the caller sleeps before the next batch to avoid
// hammering the rate limiter back-to-back.
let mut consecutive_stalls = 0u32;
// While the server is silent, re-report the current wait every few
// seconds so the UI shows the wait climbing instead of a stale
// average (the average only moves when a message actually arrives).
const STALL_REPORT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5);
loop {
let item =
match tokio::time::timeout(STALL_REPORT_INTERVAL, body_stream.try_next()).await {
Ok(item) => item
.map_err(|e| raise_error!(format!("{:#?}", e), classify_imap_error(&e)))?,
Err(_) => {
if token.is_cancelled() {
tracing::info!("Account {}: UID fetch stream interrupted.", account_id);
return Err(raise_error!(
"Stream cancelled".into(),
ErrorCode::InternalError
));
}
let stall_secs = last_recv.elapsed().as_secs_f64();
consecutive_stalls += 1;
if let Some(progress) = progress {
progress(count, None, Some(stall_secs))?;
}
continue;
}
};
let Some(fetch) = item else { break };
if token.is_cancelled() {
tracing::info!("Account {}: UID fetch stream interrupted.", account_id);
return Err(raise_error!(
@@ -489,10 +575,29 @@ impl ImapExecutor {
ErrorCode::InternalError
));
}
let now = std::time::Instant::now();
consecutive_stalls = 0;
last_recv = now;
recv_times.push_back(now);
if recv_times.len() > 10 {
recv_times.pop_front();
}
let avg_secs = if recv_times.len() >= 2 {
let span = recv_times
.back()
.unwrap()
.duration_since(*recv_times.front().unwrap());
Some(span.as_secs_f64() / (recv_times.len() as f64 - 1.0))
} else {
None
};
extract_envelope_and_store_it(fetch, account_id, mailbox_id).await?;
count += 1;
if let Some(progress) = progress {
progress(count, avg_secs, None)?;
}
}
Ok(count)
Ok((count, consecutive_stalls >= 2))
}
/// Fetches the raw RFC822 body of a single message by UID.
@@ -588,6 +693,46 @@ impl ImapExecutor {
pub const DEFAULT_BATCH_SIZE: u32 = 30;
pub const DEFAULT_MAX_EMAIL_SIZE: u64 = 100 * 1024 * 1024;
/// Average seconds between consecutive messages above which the IMAP server is
/// considered slow (a healthy server responds in milliseconds).
const SLOW_SERVER_THRESHOLD_SECS: f64 = 3.0;
/// Seconds of silence before the UI is told the server is stalling on the
/// current message (lower than SLOW_SERVER_THRESHOLD so the message flips to
/// "waiting" promptly once the server stops feeding).
const STALL_REPORT_THRESHOLD_SECS: f64 = 3.0;
/// Silence beyond this is treated as likely rate limiting (throttling), not
/// just slowness — the UI then explains what Bichon is doing about it.
const RATE_LIMIT_THRESHOLD_SECS: f64 = 10.0;
/// Returns a user-facing hint when the IMAP server is feeding messages slowly
/// or has gone silent, so the UI can explain "this is the server, not Bichon".
/// `None` when the server is responding normally or not enough messages
/// arrived to tell. `stall_secs` (server silent on the current message) takes
/// precedence over the running average.
pub fn slow_server_message(avg_secs: Option<f64>, stall_secs: Option<f64>) -> Option<String> {
if let Some(stall) = stall_secs {
if stall >= RATE_LIMIT_THRESHOLD_SECS {
return Some(format!(
"Possible IMAP rate limiting: server has been silent for {:.0}s. Bichon is pacing the download (pausing between batches) and will retry with backoff if the connection stalls.",
stall
));
}
if stall >= STALL_REPORT_THRESHOLD_SECS {
return Some(format!(
"IMAP server is slow: no response for {:.0}s while fetching the next message; download is still in progress.",
stall
));
}
}
match avg_secs {
Some(secs) if secs >= SLOW_SERVER_THRESHOLD_SECS => Some(format!(
"IMAP server is slow (avg {:.1}s between messages); download is still in progress.",
secs
)),
_ => None,
}
}
/// Compresses a sorted list of UIDs into an IMAP sequence-set string.
/// Consecutive UIDs become ranges (e.g. `1:5`), non-consecutive are
/// comma-separated (e.g. `1:5,10,12:15`).
@@ -737,10 +882,7 @@ mod test {
#[test]
fn parse_message_id_lowercase() {
let header = b"Message-Id: <foo@bar.com>\r\n";
assert_eq!(
parse_message_id_header(header),
Some("foo@bar.com".into())
);
assert_eq!(parse_message_id_header(header), Some("foo@bar.com".into()));
}
#[test]

8
crates/core/src/settings/cli.rs
View File

@@ -242,6 +242,14 @@ pub struct Settings {
)]
pub bichon_sync_concurrency: Option<u16>,
#[clap(
long,
env,
default_value = "90",
help = "IMAP socket read timeout in seconds (0 disables the timeout). Servers that throttle or burst slowly (e.g. Zoho) can pause for 30-60s between responses; keep this above the longest expected server silence so throttling surfaces as progress delay, not a failed sync."
)]
pub bichon_imap_timeout_seconds: u64,
#[clap(
long,
env,

13
crates/core/src/utils/net.rs
View File

@@ -18,7 +18,7 @@
use crate::error::code::ErrorCode;
use crate::raise_error;
use crate::settings::proxy::Proxy;
use crate::settings::{cli::SETTINGS, proxy::Proxy};
use crate::utils::tls::establish_tls_stream;
use crate::{error::BichonResult, imap::session::SessionStream};
use base64::{engine::general_purpose, Engine as _};
@@ -90,9 +90,16 @@ pub(crate) async fn establish_tcp_connection_with_timeout(
let tcp_stream = connect_with_optional_proxy(use_proxy, address).await?;
let mut timeout_stream = TimeoutStream::new(tcp_stream);
// Set read and write timeouts
// Set read and write timeouts. The read timeout bounds how long the sync
// task blocks waiting for a slow IMAP server between commands; a hung
// server therefore surfaces as a network error (and retry) instead of a
// silently stuck download. 0 disables the read timeout (server decides).
let read_timeout = SETTINGS
.bichon_imap_timeout_seconds
.checked_sub(1)
.map(|seconds| Duration::from_secs(seconds.max(1)));
timeout_stream.set_write_timeout(Some(Duration::from_secs(15)));
timeout_stream.set_read_timeout(Some(Duration::from_secs(30)));
timeout_stream.set_read_timeout(read_timeout);
// Return the timeout-wrapped TCP stream as a Pin
Ok(Box::pin(timeout_stream))