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

chore: remove bb8 pool for IMAP; create a new session per operation to avoid stale connections

Browse Source
This commit is contained in:
rustmailer
2026-01-28 20:41:17 +08:00
parent d63b1e0d7c
commit 579822762f
16 changed files with 186 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-13
View File
@@ -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",
Cargo.toml
-1
View File
@@ -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"
src/main.rs
+2 -2
View File
@@ -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(())
}
src/modules/account/migration.rs
+1 -2
View File
@@ -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?;
src/modules/cache/imap/sync/flow.rs
Vendored
+38 -24
View File
@@ -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<usize> {
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!(
src/modules/cache/imap/sync/mod.rs
Vendored
+15 -2
View File
@@ -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?;
src/modules/cache/imap/sync/rebuild.rs
Vendored
-2
View File
@@ -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);
src/modules/cache/imap/sync/sync_folders.rs
Vendored
+8 -7
View File
@@ -16,7 +16,6 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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<Vec<MailBox>> {
pub async fn get_sync_folders(
account: &AccountModel,
session: &mut Session<Box<dyn SessionStream>>,
) -> BichonResult<Vec<MailBox>> {
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<Vec<MailBo
), ErrorCode::ImapUnexpectedResult));
}
}
convert_names_to_mailboxes(account.id, matched_mailboxes).await
convert_names_to_mailboxes(account.id, session, matched_mailboxes).await
}
pub async fn detect_mailbox_changes(
src/modules/context/executors.rs
+7 -43
View File
@@ -18,73 +18,37 @@
use crate::modules::account::migration::AccountType;
use crate::modules::context::Initialize;
use crate::modules::error::code::ErrorCode;
use crate::raise_error;
use crate::{
modules::{
account::migration::AccountModel,
context::controller::SYNC_CONTROLLER,
error::BichonResult,
imap::{executor::ImapExecutor, pool::build_imap_pool},
account::migration::AccountModel, context::controller::SYNC_CONTROLLER, error::BichonResult,
},
utc_now,
};
use dashmap::DashMap;
use std::sync::{Arc, LazyLock};
use std::sync::LazyLock;
use tracing::info;
pub static MAIL_CONTEXT: LazyLock<EmailClientExecutors> = LazyLock::new(EmailClientExecutors::new);
pub static BICHON_CONTEXT: LazyLock<BichonContext> = LazyLock::new(BichonContext::new);
pub struct EmailClientExecutors {
pub struct BichonContext {
start_at: i64,
imap: DashMap<u64, Arc<ImapExecutor>>,
}
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<Arc<ImapExecutor>> {
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<AccountModel> = accounts
src/modules/context/status.rs
+3 -3
View File
@@ -17,7 +17,7 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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(),
}
src/modules/imap/executor.rs
+66 -67
View File
@@ -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<ImapConnectionManager>,
}
pub struct ImapExecutor;
impl ImapExecutor {
pub fn new(account_id: u64, pool: Pool<ImapConnectionManager>) -> Self {
Self { account_id, pool }
}
pub async fn list_all_mailboxes(&self) -> BichonResult<Vec<Name>> {
let mut session = self.get_connection().await?;
pub async fn list_all_mailboxes(
session: &mut Session<Box<dyn SessionStream>>,
) -> BichonResult<Vec<Name>> {
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<Mailbox> {
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<HashSet<u32>> {
let mut session = self.get_connection().await?;
pub async fn uid_search(
session: &mut Session<Box<dyn SessionStream>>,
mailbox_name: &str,
query: &str,
) -> BichonResult<HashSet<u32>> {
session
.examine(mailbox_name)
.await
@@ -80,13 +70,12 @@ impl ImapExecutor {
}
pub async fn append(
&self,
session: &mut Session<Box<dyn SessionStream>>,
mailbox_name: impl AsRef<str>,
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<Box<dyn SessionStream>>,
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<Box<dyn SessionStream>>,
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<Box<dyn SessionStream>>,
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<bb8::PooledConnection<'_, ImapConnectionManager>> {
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<bb8::PooledConnection<'_, ImapConnectionManager>> {
// 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<Session<Box<dyn SessionStream>>> {
ImapConnectionManager::build(account_id).await
}
}
src/modules/imap/manager.rs
+12 -26
View File
@@ -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<AccountModel> {
// Fetch the account entity in non-test environment
AccountModel::get(self.account_id).await
}
async fn create_client(&self, account: &AccountModel) -> BichonResult<Client> {
async fn create_client(account: &AccountModel) -> BichonResult<Client> {
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<Session<Box<dyn SessionStream>>> {
@@ -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<Session<Box<dyn SessionStream>>> {
let account = self.fetch_account().await?;
let client = match self.create_client(&account).await {
pub async fn build(account_id: u64) -> BichonResult<Session<Box<dyn SessionStream>>> {
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<String> = 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;
src/modules/imap/mod.rs
-1
View File
@@ -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)]
src/modules/imap/pool.rs
-60
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<Box<dyn SessionStream>>;
type Error = BichonError;
async fn connect(&self) -> BichonResult<Self::Connection> {
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<Pool<ImapConnectionManager>> {
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)
}
src/modules/mailbox/list.rs
+22 -37
View File
@@ -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<Vec<MailBox>> {
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<Vec<MailBox>> {
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<Box<dyn SessionStream>>,
names: impl IntoIterator<Item = &Name>,
) -> BichonResult<Vec<MailBox>> {
// 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<Result<MailBox, BichonError>> =
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)
src/modules/message/append.rs
+12 -6
View File
@@ -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<u64>) -> 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<u64>) -> 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<u64>) -> BichonRes
);
}
session.logout().await.ok();
Ok(())
}