From 579822762f0c655018b1230598de76fb12241cba Mon Sep 17 00:00:00 2001 From: rustmailer Date: Wed, 28 Jan 2026 20:41:17 +0800 Subject: [PATCH] chore: remove bb8 pool for IMAP; create a new session per operation to avoid stale connections --- Cargo.lock | 13 -- Cargo.toml | 1 - src/main.rs | 4 +- src/modules/account/migration.rs | 3 +- src/modules/cache/imap/sync/flow.rs | 62 +++++---- src/modules/cache/imap/sync/mod.rs | 17 ++- src/modules/cache/imap/sync/rebuild.rs | 2 - src/modules/cache/imap/sync/sync_folders.rs | 15 +-- src/modules/context/executors.rs | 50 ++------ src/modules/context/status.rs | 6 +- src/modules/imap/executor.rs | 133 ++++++++++---------- src/modules/imap/manager.rs | 38 ++---- src/modules/imap/mod.rs | 1 - src/modules/imap/pool.rs | 60 --------- src/modules/mailbox/list.rs | 59 ++++----- src/modules/message/append.rs | 18 ++- 16 files changed, 186 insertions(+), 296 deletions(-) delete mode 100644 src/modules/imap/pool.rs diff --git a/Cargo.lock b/Cargo.lock index f641cc9..98f3328 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -430,18 +430,6 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "bb8" -version = "0.9.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "457d7ed3f888dfd2c7af56d4975cade43c622f74bdcddfed6d4352f57acc6310" -dependencies = [ - "futures-util", - "parking_lot", - "portable-atomic", - "tokio", -] - [[package]] name = "bichon" version = "0.3.7" @@ -450,7 +438,6 @@ dependencies = [ "async-imap", "autoconfig", "base64 0.22.1", - "bb8", "bytes 1.11.0", "cacache", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 220f125..46392b7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -93,7 +93,6 @@ webpki-roots = "1.0.5" rustls = { version = "0.23.36", default-features = false, features = ["ring"] } rustls-pki-types = "1.14.0" tokio-io-timeout = "1.2.1" -bb8 = "0.9.1" semver = "1.0.27" governor = "0.10.4" lru = "0.16.3" diff --git a/src/main.rs b/src/main.rs index 104f084..1485018 100644 --- a/src/main.rs +++ b/src/main.rs @@ -20,7 +20,7 @@ use bichon::{ bichon_version, modules::{ common::rustls::RustMailerTls, - context::{executors::EmailClientExecutors, Initialize}, + context::{executors::BichonContext, Initialize}, error::BichonResult, logger, rest::start_http_server, @@ -71,7 +71,7 @@ async fn initialize() -> BichonResult<()> { DataDirManager::initialize().await?; UserManager::initialize().await?; RustMailerTls::initialize().await?; - EmailClientExecutors::initialize().await?; + BichonContext::initialize().await?; PeriodicTasks::start_background_tasks(); Ok(()) } diff --git a/src/modules/account/migration.rs b/src/modules/account/migration.rs index 6840084..2693cc3 100644 --- a/src/modules/account/migration.rs +++ b/src/modules/account/migration.rs @@ -47,7 +47,6 @@ use crate::modules::account::payload::AccountUpdateRequest; use crate::modules::account::payload::MinimalAccount; use crate::modules::cache::imap::task::SYNC_TASKS; use crate::modules::context::controller::SYNC_CONTROLLER; -use crate::modules::context::executors::MAIL_CONTEXT; use crate::modules::database::count_by_unique_secondary_key_impl; use crate::modules::database::delete_impl; use crate::modules::database::manager::DB_MANAGER; @@ -297,7 +296,7 @@ impl AccountV3 { if matches!(account.account_type, AccountType::IMAP) { SYNC_TASKS.stop(account.id).await?; AccountRunningState::delete(account.id).await?; - MAIL_CONTEXT.clean_account(account.id).await?; + //BICHON_CONTEXT.clean_account(account.id).await?; } OAuth2AccessToken::try_delete(account.id).await?; UserModel::cleanup_account(account.id).await?; diff --git a/src/modules/cache/imap/sync/flow.rs b/src/modules/cache/imap/sync/flow.rs index a2a4625..92ca8d3 100644 --- a/src/modules/cache/imap/sync/flow.rs +++ b/src/modules/cache/imap/sync/flow.rs @@ -30,8 +30,8 @@ use crate::{ }, SEMAPHORE, }, - context::executors::MAIL_CONTEXT, error::{code::ErrorCode, BichonError, BichonResult}, + imap::executor::ImapExecutor, indexer::manager::ENVELOPE_INDEX_MANAGER, }, raise_error, @@ -55,16 +55,15 @@ pub async fn fetch_and_save_by_date( direction: FetchDirection, ) -> BichonResult { let account_id = account.id; - let executor = MAIL_CONTEXT.imap(account_id).await?; + let mut session = ImapExecutor::create_connection(account_id).await?; let search_criteria = match direction { FetchDirection::Since => format!("SINCE {date}"), FetchDirection::Before => format!("BEFORE {date}"), }; - let uid_list = executor - .uid_search(&mailbox.encoded_name(), &search_criteria) - .await?; + let uid_list = + ImapExecutor::uid_search(&mut session, &mailbox.encoded_name(), &search_criteria).await?; let len = uid_list.len(); if len == 0 { @@ -109,12 +108,18 @@ pub async fn fetch_and_save_by_date( (index + 1) as u32, ) .await?; - let executor = MAIL_CONTEXT.imap(account_id).await?; + // Fetch metadata for the current batch of UIDs - executor - .uid_batch_retrieve_emails(account_id, mailbox.id, &batch, &mailbox.encoded_name()) - .await?; + ImapExecutor::uid_batch_retrieve_emails( + &mut session, + account_id, + mailbox.id, + &batch, + &mailbox.encoded_name(), + ) + .await?; } + session.logout().await.ok(); Ok(len) } @@ -153,26 +158,29 @@ pub async fn fetch_and_save_full_mailbox( "Starting full mailbox sync for '{}', total={}, limit={:?}, batches={}, desc={}", mailbox.name, total, folder_limit, total_batches, desc ); + + let mut session = ImapExecutor::create_connection(account_id).await?; + for page in 1..=total_batches { AccountRunningState::set_current_sync_batch_number(account_id, mailbox.name.clone(), page) .await?; - let executor = MAIL_CONTEXT.imap(account_id).await?; - let count = executor - .batch_retrieve_emails( - account_id, - mailbox_id, - page as u64, - page_size as u64, - &mailbox.encoded_name(), - desc, - ) - .await?; + let count = ImapExecutor::batch_retrieve_emails( + &mut session, + account_id, + mailbox_id, + page as u64, + page_size as u64, + &mailbox.encoded_name(), + desc, + ) + .await?; inserted_count += count; info!( "Batch insertion completed for mailbox: {}, current page: {}, inserted count: {}", &mailbox.name, page, count ); } + session.logout().await.ok(); Ok(inserted_count) } @@ -423,16 +431,22 @@ async fn perform_incremental_sync( .await?; match local_max_uid { Some(max_uid) => { - let executor = MAIL_CONTEXT.imap(account.id).await?; + let mut session = ImapExecutor::create_connection(account.id).await?; let before_date = account .date_before .as_ref() .map(|r| r.calculate_date()) .transpose()?; - executor - .fetch_new_mail(account, local_mailbox, max_uid + 1, before_date.as_deref()) - .await?; + ImapExecutor::fetch_new_mail( + &mut session, + account, + local_mailbox, + max_uid + 1, + before_date.as_deref(), + ) + .await?; + session.logout().await.ok(); } None => { info!( diff --git a/src/modules/cache/imap/sync/mod.rs b/src/modules/cache/imap/sync/mod.rs index d1873a7..0729efc 100644 --- a/src/modules/cache/imap/sync/mod.rs +++ b/src/modules/cache/imap/sync/mod.rs @@ -25,6 +25,7 @@ use crate::{ }, cache::imap::{mailbox::MailBox, sync::flow::FetchDirection}, error::BichonResult, + imap::executor::ImapExecutor, }, utc_now, }; @@ -33,7 +34,7 @@ use rebuild::{rebuild_cache, rebuild_cache_by_date}; use std::time::Instant; use sync_folders::get_sync_folders; use sync_type::{determine_sync_type, SyncType}; -use tracing::debug; +use tracing::{debug, warn}; pub mod flow; pub mod rebuild; @@ -48,7 +49,19 @@ pub async fn execute_imap_sync(account: &AccountModel) -> BichonResult<()> { if matches!(sync_type, SyncType::SkipSync) { return Ok(()); } - let remote_mailboxes = get_sync_folders(account).await?; + let mut session = ImapExecutor::create_connection(account_id).await?; + let remote_mailboxes = match get_sync_folders(account, &mut session).await { + Ok(mailboxes) => mailboxes, + Err(err) => { + warn!( + account_id = account.id, + error = %err, + "Failed to get sync folders, logging out and skipping this account" + ); + return Ok(()); + } + }; + session.logout().await.ok(); if matches!(sync_type, SyncType::InitialSync) { AccountRunningState::add(account.id).await?; // AccountRunningState::set_initial_sync_start(account_id).await?; diff --git a/src/modules/cache/imap/sync/rebuild.rs b/src/modules/cache/imap/sync/rebuild.rs index d785864..c0a125e 100644 --- a/src/modules/cache/imap/sync/rebuild.rs +++ b/src/modules/cache/imap/sync/rebuild.rs @@ -121,7 +121,6 @@ pub async fn rebuild_cache_by_date( MailBox::batch_insert(remote_mailboxes).await?; let mut handles = Vec::new(); - let local_semaphore = Arc::new(Semaphore::new(DEFAULT_MAX_CONCURRENT_PER_ACCOUNT)); for mailbox in remote_mailboxes { @@ -164,7 +163,6 @@ pub async fn rebuild_cache_by_date( tokio::spawn(async move { let _global_permit = global_permit; let _local_permit = local_permit; - fetch_and_save_by_date(&account, date.as_str(), &mailbox, direction).await }); handles.push(handle); diff --git a/src/modules/cache/imap/sync/sync_folders.rs b/src/modules/cache/imap/sync/sync_folders.rs index e0a9345..b4fa28e 100644 --- a/src/modules/cache/imap/sync/sync_folders.rs +++ b/src/modules/cache/imap/sync/sync_folders.rs @@ -16,7 +16,6 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - use std::collections::BTreeSet; use crate::{ @@ -24,19 +23,21 @@ use crate::{ modules::{ account::migration::{AccountModel, AccountType}, cache::imap::mailbox::{AttributeEnum, MailBox}, - context::executors::MAIL_CONTEXT, error::{code::ErrorCode, BichonResult}, + imap::{executor::ImapExecutor, session::SessionStream}, mailbox::list::convert_names_to_mailboxes, }, raise_error, }; -use async_imap::types::Name; +use async_imap::{types::Name, Session}; use tracing::{debug, info, warn}; -pub async fn get_sync_folders(account: &AccountModel) -> BichonResult> { +pub async fn get_sync_folders( + account: &AccountModel, + session: &mut Session>, +) -> BichonResult> { assert_eq!(account.account_type, AccountType::IMAP); - let executor = MAIL_CONTEXT.imap(account.id).await?; - let names = executor.list_all_mailboxes().await?; + let names = ImapExecutor::list_all_mailboxes(session).await?; if names.is_empty() { warn!( "Account {}: No mailboxes returned from IMAP server.", @@ -121,7 +122,7 @@ pub async fn get_sync_folders(account: &AccountModel) -> BichonResult = LazyLock::new(EmailClientExecutors::new); +pub static BICHON_CONTEXT: LazyLock = LazyLock::new(BichonContext::new); -pub struct EmailClientExecutors { +pub struct BichonContext { start_at: i64, - imap: DashMap>, } -impl Initialize for EmailClientExecutors { +impl Initialize for BichonContext { async fn initialize() -> BichonResult<()> { - MAIL_CONTEXT.start_account_syncers().await + BICHON_CONTEXT.start_account_syncers().await } } -impl EmailClientExecutors { +impl BichonContext { pub fn new() -> Self { Self { start_at: utc_now!(), - imap: DashMap::new(), } } pub fn uptime_ms(&self) -> i64 { utc_now!() - self.start_at } - pub async fn imap(&self, account_id: u64) -> BichonResult> { - if let Some(executor) = self.imap.get(&account_id) { - return Ok(executor.value().clone()); - } - - let pool = build_imap_pool(account_id).await?; - let new_executor = Arc::new(ImapExecutor::new(account_id, pool)); - - match self.imap.try_entry(account_id) { - Some(dashmap::mapref::entry::Entry::Occupied(entry)) => Ok(entry.get().clone()), - Some(dashmap::mapref::entry::Entry::Vacant(entry)) => { - entry.insert(new_executor.clone()); - Ok(new_executor) - } - None => Err(raise_error!( - "DashMap locked".into(), - ErrorCode::InternalError - )), - } - } - - pub async fn clean_account(&self, account_id: u64) -> BichonResult<()> { - if self.imap.remove(&account_id).is_some() { - info!(account_id, "Closed IMAP pool for account"); - } - Ok(()) - } - pub async fn start_account_syncers(&self) -> BichonResult<()> { let accounts = AccountModel::list_all().await?; let active_accounts: Vec = accounts diff --git a/src/modules/context/status.rs b/src/modules/context/status.rs index 94b07e8..f44b615 100644 --- a/src/modules/context/status.rs +++ b/src/modules/context/status.rs @@ -17,7 +17,7 @@ // along with this program. If not, see . -use crate::modules::context::executors::MAIL_CONTEXT; +use crate::modules::context::executors::BICHON_CONTEXT; use chrono::Local; use poem_openapi::Object; use serde::Deserialize; @@ -40,9 +40,9 @@ pub struct BichonStatus { impl BichonStatus { pub fn get() -> Self { Self { - uptime_ms: MAIL_CONTEXT.uptime_ms(), + uptime_ms: BICHON_CONTEXT.uptime_ms(), timeago: Formatter::new() - .convert(Duration::from_millis(MAIL_CONTEXT.uptime_ms() as u64)), + .convert(Duration::from_millis(BICHON_CONTEXT.uptime_ms() as u64)), timezone: Local::now().offset().to_string(), version: env!("CARGO_PKG_VERSION").into(), } diff --git a/src/modules/imap/executor.rs b/src/modules/imap/executor.rs index e694d26..1829818 100644 --- a/src/modules/imap/executor.rs +++ b/src/modules/imap/executor.rs @@ -22,12 +22,13 @@ use crate::modules::cache::imap::mailbox::MailBox; use crate::modules::cache::imap::sync::flow::{generate_uid_sequence_hashset, DEFAULT_BATCH_SIZE}; use crate::modules::envelope::extractor::extract_envelope; use crate::modules::error::code::ErrorCode; +use crate::modules::imap::session::SessionStream; use crate::modules::indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}; use crate::modules::indexer::schema::SchemaTools; use crate::modules::{error::BichonResult, imap::manager::ImapConnectionManager}; use crate::raise_error; -use async_imap::types::{Mailbox, Name}; -use bb8::{Pool, RunError}; +use async_imap::types::Name; +use async_imap::Session; use futures::TryStreamExt; use std::collections::HashSet; use tantivy::doc; @@ -35,18 +36,12 @@ use tracing::info; const BODY_FETCH_COMMAND: &str = "(UID INTERNALDATE RFC822.SIZE BODY.PEEK[])"; -pub struct ImapExecutor { - account_id: u64, - pool: Pool, -} +pub struct ImapExecutor; impl ImapExecutor { - pub fn new(account_id: u64, pool: Pool) -> Self { - Self { account_id, pool } - } - - pub async fn list_all_mailboxes(&self) -> BichonResult> { - let mut session = self.get_connection().await?; + pub async fn list_all_mailboxes( + session: &mut Session>, + ) -> BichonResult> { let list = session .list(Some(""), Some("*")) .await @@ -58,16 +53,11 @@ impl ImapExecutor { Ok(result) } - pub async fn examine_mailbox(&self, mailbox_name: &str) -> BichonResult { - let mut session = self.get_connection().await?; - session - .examine(mailbox_name) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed)) - } - - pub async fn uid_search(&self, mailbox_name: &str, query: &str) -> BichonResult> { - let mut session = self.get_connection().await?; + pub async fn uid_search( + session: &mut Session>, + mailbox_name: &str, + query: &str, + ) -> BichonResult> { session .examine(mailbox_name) .await @@ -80,13 +70,12 @@ impl ImapExecutor { } pub async fn append( - &self, + session: &mut Session>, mailbox_name: impl AsRef, flags: Option<&str>, internaldate: Option<&str>, content: impl AsRef<[u8]>, ) -> BichonResult<()> { - let mut session = self.get_connection().await?; session .append(mailbox_name, flags, internaldate, content) .await @@ -94,7 +83,7 @@ impl ImapExecutor { } pub async fn fetch_new_mail( - &self, + session: &mut Session>, account: &AccountModel, mailbox: &MailBox, start_uid: u64, @@ -107,7 +96,7 @@ impl ImapExecutor { None => format!("UID {start_uid}:*"), }; - let uid_list = self.uid_search(&mailbox.encoded_name(), &query).await?; + let uid_list = Self::uid_search(session, &mailbox.encoded_name(), &query).await?; let len = uid_list.len(); if len == 0 { @@ -145,14 +134,20 @@ impl ImapExecutor { ) .await?; } - self.uid_batch_retrieve_emails(account.id, mailbox.id, &batch, &mailbox.encoded_name()) - .await?; + Self::uid_batch_retrieve_emails( + session, + account.id, + mailbox.id, + &batch, + &mailbox.encoded_name(), + ) + .await?; } Ok(()) } pub async fn batch_retrieve_emails( - &self, + session: &mut Session>, account_id: u64, mailbox_id: u64, page: u64, @@ -163,7 +158,6 @@ impl ImapExecutor { assert!(page > 0, "Page number must be greater than 0"); assert!(page_size > 0, "Page size must be greater than 0"); - let mut session = self.get_connection().await?; let total = session .examine(encoded_mailbox_name) .await @@ -226,13 +220,12 @@ impl ImapExecutor { } pub async fn uid_batch_retrieve_emails( - &self, + session: &mut Session>, account_id: u64, mailbox_id: u64, uid_set: &str, encoded_mailbox_name: &str, ) -> BichonResult<()> { - let mut session = self.get_connection().await?; session .examine(encoded_mailbox_name) .await @@ -260,40 +253,46 @@ impl ImapExecutor { Ok(()) } - async fn get_connection( - &self, - ) -> BichonResult> { - match self.pool.get().await { - Ok(connection) => Ok(connection), - Err(e) => match e { - RunError::User(e) => Err(e), - RunError::TimedOut => { - let state = self.pool.state(); - tracing::warn!( - "{}: connections={}, idle={}, \ - get_started={}, get_direct={}, get_waited={}, get_timed_out={}, \ - wait_time_ms={}, created={}, closed_broken={}, closed_invalid={}, \ - closed_lifetime={}, closed_idle={}", - self.account_id, - state.connections, - state.idle_connections, - state.statistics.get_started, - state.statistics.get_direct, - state.statistics.get_waited, - state.statistics.get_timed_out, - state.statistics.get_wait_time.as_millis(), - state.statistics.connections_created, - state.statistics.connections_closed_broken, - state.statistics.connections_closed_invalid, - state.statistics.connections_closed_max_lifetime, - state.statistics.connections_closed_idle_timeout, - ); - return Err(raise_error!( - "Timed out while attempting to acquire a connection from the pool".into(), - ErrorCode::ConnectionPoolTimeout - )); - } - }, - } + // async fn get_connection( + // &self, + // ) -> BichonResult> { + // match self.pool.get().await { + // Ok(connection) => Ok(connection), + // Err(e) => match e { + // RunError::User(e) => Err(e), + // RunError::TimedOut => { + // let state = self.pool.state(); + // tracing::warn!( + // "{}: connections={}, idle={}, \ + // get_started={}, get_direct={}, get_waited={}, get_timed_out={}, \ + // wait_time_ms={}, created={}, closed_broken={}, closed_invalid={}, \ + // closed_lifetime={}, closed_idle={}", + // self.account_id, + // state.connections, + // state.idle_connections, + // state.statistics.get_started, + // state.statistics.get_direct, + // state.statistics.get_waited, + // state.statistics.get_timed_out, + // state.statistics.get_wait_time.as_millis(), + // state.statistics.connections_created, + // state.statistics.connections_closed_broken, + // state.statistics.connections_closed_invalid, + // state.statistics.connections_closed_max_lifetime, + // state.statistics.connections_closed_idle_timeout, + // ); + // return Err(raise_error!( + // "Timed out while attempting to acquire a connection from the pool".into(), + // ErrorCode::ConnectionPoolTimeout + // )); + // } + // }, + // } + // } + + pub async fn create_connection( + account_id: u64, + ) -> BichonResult>> { + ImapConnectionManager::build(account_id).await } } diff --git a/src/modules/imap/manager.rs b/src/modules/imap/manager.rs index 4a5f158..4bd2c3e 100644 --- a/src/modules/imap/manager.rs +++ b/src/modules/imap/manager.rs @@ -32,22 +32,10 @@ use crate::{bichon_version, decrypt, raise_error}; use async_imap::Session; use tracing::error; -#[derive(Debug)] -pub struct ImapConnectionManager { - pub account_id: u64, -} +pub struct ImapConnectionManager; impl ImapConnectionManager { - pub fn new(account_id: u64) -> Self { - Self { account_id } - } - - pub async fn fetch_account(&self) -> BichonResult { - // Fetch the account entity in non-test environment - AccountModel::get(self.account_id).await - } - - async fn create_client(&self, account: &AccountModel) -> BichonResult { + async fn create_client(account: &AccountModel) -> BichonResult { assert_eq!(account.account_type, AccountType::IMAP); let imap = account.imap.as_ref().unwrap(); Client::connection( @@ -61,7 +49,6 @@ impl ImapConnectionManager { } async fn authenticate( - &self, client: Client, account: &AccountModel, ) -> BichonResult>> { @@ -87,7 +74,7 @@ impl ImapConnectionManager { }) } AuthType::OAuth2 => { - let record = OAuth2AccessToken::get(self.account_id).await?; + let record = OAuth2AccessToken::get(account.id).await?; let access_token = record.and_then(|r| r.access_token).ok_or_else(|| { raise_error!( "Imap auth type is OAuth2, but OAuth2 authorization is not yet complete." @@ -106,9 +93,9 @@ impl ImapConnectionManager { } } - pub async fn build(&self) -> BichonResult>> { - let account = self.fetch_account().await?; - let client = match self.create_client(&account).await { + pub async fn build(account_id: u64) -> BichonResult>> { + let account = AccountModel::get(account_id).await?; + let client = match Self::create_client(&account).await { Ok(client) => client, Err(error) => { error!( @@ -117,7 +104,7 @@ impl ImapConnectionManager { ); STATUS_DISPATCHER .append_error( - self.account_id, + account_id, format!("imap client connect error: {:#?}", error), ) .await; @@ -125,14 +112,13 @@ impl ImapConnectionManager { } }; - let mut session = match self.authenticate(client, &account).await { + let mut session = match Self::authenticate(client, &account).await { Ok(session) => session, Err(error) => { error!("Failed to authenticate IMAP session: {:#?}", error); - STATUS_DISPATCHER .append_error( - self.account_id, + account_id, format!("imap client authenticate error: {:#?}", error), ) .await; @@ -143,12 +129,12 @@ impl ImapConnectionManager { match fetch_capabilities(&mut session).await { Ok(capabilities) => { let to_save: Vec = capabilities.iter().map(capability_to_string).collect(); - AccountModel::update_capabilities(self.account_id, to_save).await?; + AccountModel::update_capabilities(account_id, to_save).await?; if let Err(error) = check_capabilities(&capabilities) { error!("Failed to check IMAP capabilities: {:#?}", error); STATUS_DISPATCHER .append_error( - self.account_id, + account_id, format!("imap client check capabilities error: {:#?}", error), ) .await; @@ -172,7 +158,7 @@ impl ImapConnectionManager { error!("Failed to fetch IMAP capabilities: {:#?}", error); STATUS_DISPATCHER .append_error( - self.account_id, + account_id, format!("imap client fetch capabilities error: {:#?}", error), ) .await; diff --git a/src/modules/imap/mod.rs b/src/modules/imap/mod.rs index 898b201..bbadd3c 100644 --- a/src/modules/imap/mod.rs +++ b/src/modules/imap/mod.rs @@ -22,7 +22,6 @@ pub mod client; pub mod executor; pub mod manager; pub mod oauth2; -pub mod pool; pub mod session; pub mod stats; #[cfg(test)] diff --git a/src/modules/imap/pool.rs b/src/modules/imap/pool.rs deleted file mode 100644 index 503a7f6..0000000 --- a/src/modules/imap/pool.rs +++ /dev/null @@ -1,60 +0,0 @@ -// -// Copyright (c) 2025 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 . - -use crate::modules::error::code::ErrorCode; -use crate::modules::error::{BichonError, BichonResult}; -use crate::modules::imap::{manager::ImapConnectionManager, session::SessionStream}; -use crate::raise_error; -use async_imap::Session; -use bb8::Pool; -use std::time::Duration; - -impl bb8::ManageConnection for ImapConnectionManager { - type Connection = Session>; - - type Error = BichonError; - - async fn connect(&self) -> BichonResult { - self.build().await - } - // call this function before using the connection - async fn is_valid(&self, conn: &mut Self::Connection) -> BichonResult<()> { - conn.noop() - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed)) - } - - fn has_broken(&self, _: &mut Self::Connection) -> bool { - false - } -} - -pub async fn build_imap_pool(account_id: u64) -> BichonResult> { - let manager = ImapConnectionManager::new(account_id); - let pool = Pool::builder() - .connection_timeout(Duration::from_secs(60)) - //.idle_timeout(Duration::from_secs(120)) - .retry_connection(true) - .queue_strategy(bb8::QueueStrategy::Fifo) - .max_size(10) - .test_on_check_out(true) - .build(manager) - .await?; - - Ok(pool) -} diff --git a/src/modules/mailbox/list.rs b/src/modules/mailbox/list.rs index 3f0e420..dc0b971 100644 --- a/src/modules/mailbox/list.rs +++ b/src/modules/mailbox/list.rs @@ -18,12 +18,14 @@ use crate::modules::account::migration::{AccountModel, AccountType}; use crate::modules::cache::imap::mailbox::{Attribute, AttributeEnum, MailBox}; -use crate::modules::context::executors::MAIL_CONTEXT; use crate::modules::error::code::ErrorCode; -use crate::modules::error::{BichonError, BichonResult}; +use crate::modules::error::BichonResult; +use crate::modules::imap::executor::ImapExecutor; +use crate::modules::imap::session::SessionStream; use crate::modules::utils::create_hash; use crate::raise_error; use async_imap::types::Name; +use async_imap::Session; pub async fn get_account_mailboxes(account_id: u64, remote: bool) -> BichonResult> { let account = AccountModel::check_account_exists(account_id).await?; @@ -42,9 +44,11 @@ pub async fn get_account_mailboxes(account_id: u64, remote: bool) -> BichonResul } pub async fn request_imap_all_mailbox_list(account_id: u64) -> BichonResult> { - let executor = MAIL_CONTEXT.imap(account_id).await?; - let names = executor.list_all_mailboxes().await?; - convert_names_to_mailboxes(account_id, names.iter()).await + let mut session = ImapExecutor::create_connection(account_id).await?; + let names = ImapExecutor::list_all_mailboxes(&mut session).await?; + let result = convert_names_to_mailboxes(account_id, &mut session, names.iter()).await?; + session.logout().await.ok(); + Ok(result) } fn contains_no_select(attributes: &[Attribute]) -> bool { @@ -55,50 +59,31 @@ fn contains_no_select(attributes: &[Attribute]) -> bool { pub async fn convert_names_to_mailboxes( account_id: u64, + session: &mut Session>, names: impl IntoIterator, ) -> BichonResult> { - // Preallocate enough space in the vector to avoid multiple reallocations - let mut tasks = Vec::new(); + let mut mailboxes = Vec::new(); - for name in names.into_iter() { - // Convert the name into a MailBox structure + for name in names { let mailbox_name = name.name().to_string(); - let mut mailbox: MailBox = name.into(); - tracing::debug!( - raw = &mailbox_name, - decoded = &mailbox.name, - "mailbox name comparison" - ); - if contains_no_select(&mailbox.attributes) { continue; } + mailbox.account_id = account_id; mailbox.id = create_hash(account_id, &mailbox.name); - let task: tokio::task::JoinHandle> = - tokio::spawn(async move { - let executor = MAIL_CONTEXT.imap(account_id).await?; - let mx = executor.examine_mailbox(mailbox_name.as_str()).await?; - // Update the mailbox status information - mailbox.exists = mx.exists; // Number of messages in the mailbox - mailbox.unseen = mx.unseen; // Number of unseen messages - mailbox.uid_next = mx.uid_next; // Next unique identifier to be assigned - mailbox.uid_validity = mx.uid_validity; // Validity of the UIDs - Ok(mailbox) - }); - tasks.push(task); - } + let mx = session + .examine(mailbox_name.as_str()) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?; + mailbox.exists = mx.exists; + mailbox.unseen = mx.unseen; + mailbox.uid_next = mx.uid_next; + mailbox.uid_validity = mx.uid_validity; - let mut mailboxes = Vec::new(); - - for task in tasks { - match task.await { - Ok(Ok(mailbox)) => mailboxes.push(mailbox), - Ok(Err(err)) => return Err(err), // Handle mailbox-level errors - Err(e) => return Err(raise_error!(format!("{:#?}", e), ErrorCode::InternalError)), // Handle task-level panics or errors - } + mailboxes.push(mailbox); } Ok(mailboxes) diff --git a/src/modules/message/append.rs b/src/modules/message/append.rs index 52914a5..0e742f5 100644 --- a/src/modules/message/append.rs +++ b/src/modules/message/append.rs @@ -2,8 +2,8 @@ use crate::{ encode_mailbox_name, modules::{ account::migration::{AccountModel, AccountType}, - context::executors::MAIL_CONTEXT, error::{code::ErrorCode, BichonResult}, + imap::executor::ImapExecutor, indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}, }, raise_error, @@ -38,10 +38,9 @@ pub async fn restore_emails(account_id: u64, message_ids: Vec) -> BichonRes ErrorCode::Incompatible )); } - let executor = MAIL_CONTEXT.imap(account.id).await?; let mut failed = Vec::new(); - + let mut session = ImapExecutor::create_connection(account_id).await?; for message_id in message_ids { let result: BichonResult<()> = async { let envelope = ENVELOPE_INDEX_MANAGER @@ -71,9 +70,14 @@ pub async fn restore_emails(account_id: u64, message_ids: Vec) -> BichonRes })?; if let Some(mailbox_name) = envelope.mailbox_name { - executor - .append(encode_mailbox_name!(&mailbox_name), None, None, &eml) - .await?; + ImapExecutor::append( + &mut session, + encode_mailbox_name!(&mailbox_name), + None, + None, + &eml, + ) + .await?; } Ok(()) @@ -100,5 +104,7 @@ pub async fn restore_emails(account_id: u64, message_ids: Vec) -> BichonRes ); } + session.logout().await.ok(); + Ok(()) }