starred/rustmailer
1
0
Fork 0
mirror of https://github.com/rustmailer/rustmailer.git synced 2026-09-07 00:01:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: Optimize the cleanup workflow.

Browse Source
This commit is contained in:
rustmailer
2026-08-27 00:45:46 +08:00
parent 905552055b
commit 7daffa09e0
7 changed files with 1346 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/cache/imap/address.rs
Vendored
+51 -92
View File
@@ -2,7 +2,7 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use std::{collections::HashSet, sync::Arc, time::Instant};
use std::{collections::HashSet, sync::Arc};
use native_db::*;
use native_model::{native_model, Model};
@@ -19,11 +19,13 @@ use crate::{
gmail::sync::envelope::GmailEnvelope, outlook::sync::envelope::OutlookEnvelope,
},
},
database::{batch_delete_impl, filter_by_secondary_key_impl, manager::DB_MANAGER},
error::{code::ErrorCode, RustMailerResult},
database::{
enqueue_delete_secondary_impl, filter_by_secondary_key_impl, manager::DB_MANAGER,
safe_delete::RowFilter,
},
error::RustMailerResult,
utils::envelope_hash,
},
raise_error,
};
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
@@ -77,35 +79,20 @@ impl AddressEntity {
}
pub async fn clean_account(account_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<AddressEntity> = rw
.scan()
.secondary(AddressEntityKey::account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<AddressEntity> = Arc::new(|_: &AddressEntity| true);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
AddressEntityKey::account_id,
account_id,
filter,
BATCH_SIZE,
format!("AddressEntity::clean_account account_id={}", account_id),
)?;
info!(
"Finished deleting address entities for account_id={} total_deleted={} in {:?}",
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of address entities for account_id={}",
account_id
);
Ok(())
}
@@ -115,9 +102,7 @@ impl AddressEntity {
mailbox_id: u64,
to_delete_uid: &[u32],
) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
const BATCH_SIZE: usize = 50;
let to_delete_set: HashSet<u64> = to_delete_uid
.iter()
@@ -125,73 +110,47 @@ impl AddressEntity {
.collect();
let to_delete_set = Arc::new(to_delete_set);
loop {
let to_delete_set = to_delete_set.clone();
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<AddressEntity> = rw
.scan()
.secondary(AddressEntityKey::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &AddressEntity| {
e.account_id == account_id && to_delete_set.contains(&e.envelope_hash)
})
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
let filter: RowFilter<AddressEntity> = Arc::new(move |e: &AddressEntity| {
e.account_id == account_id && to_delete_set.contains(&e.envelope_hash)
});
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
AddressEntityKey::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"AddressEntity::clean_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
info!(
"Finished deleting address entities for mailbox_id={} account_id={} total_deleted={} in {:?}",
mailbox_id,
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of address entities for mailbox_id={} account_id={}",
mailbox_id, account_id
);
Ok(())
}
pub async fn clean_mailbox_envelopes(account_id: u64, mailbox_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<AddressEntity> = rw
.scan()
.secondary(AddressEntityKey::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &AddressEntity| e.account_id == account_id)
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<AddressEntity> =
Arc::new(move |e: &AddressEntity| e.account_id == account_id);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
AddressEntityKey::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"AddressEntity::clean_mailbox_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
info!(
"Finished deleting address entities for mailbox_id={} account_id={} total_deleted={} in {:?}",
mailbox_id,
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of address entities for mailbox_id={} account_id={}",
mailbox_id, account_id
);
Ok(())
}
src/modules/cache/imap/migration.rs
Vendored
+52 -91
View File
@@ -2,9 +2,8 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use std::{collections::HashSet, sync::Arc, time::Instant};
use std::{collections::HashSet, sync::Arc};
use itertools::Itertools;
use native_db::*;
use native_model::{native_model, Model};
use poem_openapi::Object;
@@ -27,8 +26,9 @@ use crate::{
},
common::Addr,
database::{
batch_delete_impl, filter_by_secondary_key_impl, manager::DB_MANAGER,
paginate_secondary_scan_impl, secondary_find_impl, update_impl, with_transaction,
enqueue_delete_secondary_impl, filter_by_secondary_key_impl, manager::DB_MANAGER,
paginate_secondary_scan_impl, safe_delete::RowFilter, secondary_find_impl, update_impl,
with_transaction,
},
error::{code::ErrorCode, RustMailerResult},
imap::section::{EmailBodyPart, ImapAttachment},
@@ -448,37 +448,24 @@ impl EmailEnvelopeV3 {
}
pub async fn clean_mailbox_envelopes(account_id: u64, mailbox_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailEnvelopeV3> = rw
.scan()
.secondary(EmailEnvelopeV3Key::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &EmailEnvelopeV3| e.account_id == account_id)
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<EmailEnvelopeV3> =
Arc::new(move |e: &EmailEnvelopeV3| e.account_id == account_id);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
EmailEnvelopeV3Key::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"EmailEnvelopeV3::clean_mailbox_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
info!(
"Finished deleting envelopes for mailbox_id={} account_id={} total_deleted={} in {:?}",
mailbox_id,
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of envelopes for mailbox_id={} account_id={}",
mailbox_id, account_id
);
Ok(())
}
@@ -488,75 +475,49 @@ impl EmailEnvelopeV3 {
mailbox_id: u64,
to_delete_uid: &[u32],
) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
const BATCH_SIZE: usize = 50;
let to_delete_set: HashSet<u32> = to_delete_uid.iter().copied().collect();
let to_delete_set = Arc::new(to_delete_set);
loop {
let to_delete_set = to_delete_set.clone();
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailEnvelopeV3> = rw
.scan()
.secondary(EmailEnvelopeV3Key::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok)
.filter(|e: &EmailEnvelopeV3| {
e.account_id == account_id && to_delete_set.contains(&e.uid)
})
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
let filter: RowFilter<EmailEnvelopeV3> = Arc::new(move |e: &EmailEnvelopeV3| {
e.account_id == account_id && to_delete_set.contains(&e.uid)
});
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
EmailEnvelopeV3Key::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"EmailEnvelopeV3::clean_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
info!(
"Finished deleting envelopes for account_id={} total_deleted={} in {:?}",
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of envelopes for account_id={} mailbox_id={}",
account_id, mailbox_id
);
Ok(())
}
pub async fn clean_account(account_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailEnvelopeV3> = rw
.scan()
.secondary(EmailEnvelopeV3Key::account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.take(BATCH_SIZE)
.try_collect()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<EmailEnvelopeV3> = Arc::new(|_: &EmailEnvelopeV3| true);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
EmailEnvelopeV3Key::account_id,
account_id,
filter,
BATCH_SIZE,
format!(
"EmailEnvelopeV3::clean_account account_id={}",
account_id
),
)?;
info!(
"Finished deleting envelopes for account_id={} total_deleted={} in {:?}",
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of envelopes for account_id={}",
account_id
);
Ok(())
}
src/modules/cache/imap/minimal.rs
Vendored
+48 -90
View File
@@ -2,7 +2,7 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use std::{collections::HashSet, sync::Arc, time::Instant};
use std::{collections::HashSet, sync::Arc};
use native_db::*;
use native_model::{native_model, Model};
@@ -13,8 +13,8 @@ use crate::{
modules::{
cache::imap::{manager::EnvelopeFlagsManager, migration::EmailEnvelopeV3},
database::{
batch_delete_impl, batch_insert_impl, filter_by_secondary_key_impl,
manager::DB_MANAGER, update_impl,
batch_insert_impl, enqueue_delete_secondary_impl, filter_by_secondary_key_impl,
manager::DB_MANAGER, safe_delete::RowFilter, update_impl,
},
error::{code::ErrorCode, RustMailerResult},
utils::envelope_hash,
@@ -68,37 +68,24 @@ impl MinimalEnvelope {
}
pub async fn clean_mailbox_envelopes(account_id: u64, mailbox_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<MinimalEnvelope> = rw
.scan()
.secondary(MinimalEnvelopeKey::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &MinimalEnvelope| e.account_id == account_id)
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<MinimalEnvelope> =
Arc::new(move |e: &MinimalEnvelope| e.account_id == account_id);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
MinimalEnvelopeKey::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"MinimalEnvelope::clean_mailbox_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
info!(
"Finished deleting envelopes for mailbox_id={} account_id={} total_deleted={} in {:?}",
mailbox_id,
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of envelopes for mailbox_id={} account_id={}",
mailbox_id, account_id
);
Ok(())
}
@@ -108,41 +95,27 @@ impl MinimalEnvelope {
mailbox_id: u64,
to_delete_uid: &[u32],
) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
const BATCH_SIZE: usize = 50;
let to_delete_set: HashSet<u32> = to_delete_uid.iter().copied().collect();
let to_delete_set = Arc::new(to_delete_set);
loop {
let to_delete_set = to_delete_set.clone();
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<MinimalEnvelope> = rw
.scan()
.secondary(MinimalEnvelopeKey::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok)
.filter(|e: &MinimalEnvelope| {
e.account_id == account_id && to_delete_set.contains(&e.uid)
})
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
let filter: RowFilter<MinimalEnvelope> = Arc::new(move |e: &MinimalEnvelope| {
e.account_id == account_id && to_delete_set.contains(&e.uid)
});
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
MinimalEnvelopeKey::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"MinimalEnvelope::clean_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
info!(
"Finished deleting minimal envelopes for account_id={} total_deleted={} in {:?}",
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of minimal envelopes for account_id={} mailbox_id={}",
account_id, mailbox_id
);
Ok(())
}
@@ -186,35 +159,20 @@ impl MinimalEnvelope {
}
pub async fn clean_account(account_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<MinimalEnvelope> = rw
.scan()
.secondary(MinimalEnvelopeKey::account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<MinimalEnvelope> = Arc::new(|_: &MinimalEnvelope| true);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
MinimalEnvelopeKey::account_id,
account_id,
filter,
BATCH_SIZE,
format!("MinimalEnvelope::clean_account account_id={}", account_id),
)?;
info!(
"Finished deleting envelopes for account_id={} total_deleted={} in {:?}",
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of envelopes for account_id={}",
account_id
);
Ok(())
}
src/modules/cache/imap/thread.rs
Vendored
+46 -76
View File
@@ -2,10 +2,9 @@
// Licensed under RustMailer License Agreement v1.0
// Unauthorized copying, modification, or distribution is prohibited.
use std::{collections::HashSet, sync::Arc, time::Instant};
use std::{collections::HashSet, sync::Arc};
use futures::future::join_all;
use itertools::Itertools;
use native_db::*;
use native_model::{native_model, Model};
use poem_openapi::Object;
@@ -23,7 +22,10 @@ use crate::{
outlook::sync::envelope::OutlookEnvelope,
},
},
database::{batch_delete_impl, manager::DB_MANAGER, paginate_secondary_scan_impl},
database::{
enqueue_delete_secondary_impl, manager::DB_MANAGER, paginate_secondary_scan_impl,
safe_delete::RowFilter,
},
error::{code::ErrorCode, RustMailerResult},
rest::response::DataPage,
utils::envelope_hash,
@@ -74,51 +76,34 @@ impl EmailThread {
}
pub async fn clean_mailbox_envelopes(account_id: u64, mailbox_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailThread> = rw
.scan()
.secondary(EmailThreadKey::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &EmailThread| e.account_id == account_id)
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<EmailThread> =
Arc::new(move |e: &EmailThread| e.account_id == account_id);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
EmailThreadKey::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"EmailThread::clean_mailbox_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
Ok(())
}
pub async fn clean_account(account_id: u64) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
loop {
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailThread> = rw
.scan()
.secondary(EmailThreadKey::account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(account_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.take(BATCH_SIZE)
.try_collect()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(to_delete)
})
.await?;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
const BATCH_SIZE: usize = 50;
let filter: RowFilter<EmailThread> = Arc::new(|_: &EmailThread| true);
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
EmailThreadKey::account_id,
account_id,
filter,
BATCH_SIZE,
format!("EmailThread::clean_account account_id={}", account_id),
)?;
Ok(())
}
@@ -127,9 +112,7 @@ impl EmailThread {
mailbox_id: u64,
to_delete_uid: &[u32],
) -> RustMailerResult<()> {
const BATCH_SIZE: usize = 200;
let mut total_deleted = 0usize;
let start_time = Instant::now();
const BATCH_SIZE: usize = 50;
let to_delete_set: HashSet<u64> = to_delete_uid
.iter()
@@ -137,37 +120,24 @@ impl EmailThread {
.collect();
let to_delete_set = Arc::new(to_delete_set);
loop {
let to_delete_set = to_delete_set.clone();
let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
let to_delete: Vec<EmailThread> = rw
.scan()
.secondary(EmailThreadKey::mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(mailbox_id)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.filter_map(Result::ok) // filter only Ok values
.filter(|e: &EmailThread| {
e.account_id == account_id && to_delete_set.contains(&e.envelope_id)
})
.take(BATCH_SIZE)
.collect();
Ok(to_delete)
})
.await?;
total_deleted += deleted;
// If this batch is empty, break the loop
if deleted == 0 {
break;
}
}
let filter: RowFilter<EmailThread> = Arc::new(move |e: &EmailThread| {
e.account_id == account_id && to_delete_set.contains(&e.envelope_id)
});
enqueue_delete_secondary_impl(
DB_MANAGER.envelope_db(),
EmailThreadKey::mailbox_id,
mailbox_id,
filter,
BATCH_SIZE,
format!(
"EmailThread::clean_envelopes account_id={} mailbox_id={}",
account_id, mailbox_id
),
)?;
info!(
"Finished deleting thread entities for mailbox_id={} account_id={} total_deleted={} in {:?}",
mailbox_id,
account_id,
total_deleted,
start_time.elapsed()
"Enqueued deletion of thread entities for mailbox_id={} account_id={}",
mailbox_id, account_id
);
Ok(())
}
src/modules/database/mod.rs
+3
View File
@@ -29,6 +29,9 @@ use transaction::RwTransaction;
use super::error::code::ErrorCode;
pub mod backup;
pub mod manager;
#[allow(dead_code)] // synchronous helpers kept for direct callers / tests; cleanup uses the queue
pub mod safe_delete;
pub use safe_delete::enqueue_delete_secondary_impl;
pub mod snapshot;
#[cfg(test)]
mod tests;
src/modules/database/safe_delete.rs
+1145
View File
File diff suppressed because it is too large Load Diff
src/modules/settings/cli.rs
+1
View File
@@ -486,6 +486,7 @@ impl Settings {
rustmailer_metadata_snapshot_interval_secs: 900,
rustmailer_oauth2_success_redirect: None,
rustmailer_sync_concurrency: Some(5),
rustmailer_watchdog_timeout_secs: 0,
}
}
}