From 2250fe995e794beda0b462fb7756ed4a95b84ea3 Mon Sep 17 00:00:00 2001 From: rustmailer Date: Mon, 31 Aug 2026 14:44:11 +0800 Subject: [PATCH] fix(cache): use point query instead of scan when deleting local cache --- src/modules/cache/imap/address.rs | 48 ++++----- src/modules/cache/imap/manager.rs | 9 +- src/modules/cache/imap/migration.rs | 48 +++++---- src/modules/cache/imap/minimal.rs | 46 ++++---- src/modules/cache/imap/thread.rs | 43 ++++---- src/modules/database/safe_delete.rs | 158 ++++++++++++++++++++++++++++ 6 files changed, 256 insertions(+), 96 deletions(-) diff --git a/src/modules/cache/imap/address.rs b/src/modules/cache/imap/address.rs index c55001c..c6ee7a2 100644 --- a/src/modules/cache/imap/address.rs +++ b/src/modules/cache/imap/address.rs @@ -2,7 +2,7 @@ // Licensed under RustMailer License Agreement v1.0 // Unauthorized copying, modification, or distribution is prohibited. -use std::{collections::HashSet, sync::Arc}; +use std::sync::Arc; use native_db::*; use native_model::{native_model, Model}; @@ -20,12 +20,13 @@ use crate::{ }, }, database::{ - enqueue_delete_secondary_impl, filter_by_secondary_key_impl, manager::DB_MANAGER, - safe_delete::RowFilter, + batch_delete_impl, enqueue_delete_secondary_impl, filter_by_secondary_key_impl, + manager::DB_MANAGER, safe_delete::RowFilter, }, - error::RustMailerResult, + error::{code::ErrorCode, RustMailerResult}, utils::envelope_hash, }, + raise_error, }; #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)] @@ -102,32 +103,31 @@ impl AddressEntity { mailbox_id: u64, to_delete_uid: &[u32], ) -> RustMailerResult<()> { - const BATCH_SIZE: usize = 50; - - let to_delete_set: HashSet = to_delete_uid + let hashes: Vec = to_delete_uid .iter() .map(|uid| envelope_hash(account_id, mailbox_id, *uid)) .collect(); - let to_delete_set = Arc::new(to_delete_set); - let filter: RowFilter = 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 - ), - )?; + let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| { + let mut to_delete = Vec::new(); + for hash in hashes { + let entities: Vec = rw + .scan() + .secondary(AddressEntityKey::envelope_hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + .start_with(hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + .filter_map(Result::ok) + .collect(); + to_delete.extend(entities); + } + Ok(to_delete) + }) + .await?; info!( - "Enqueued deletion of address entities for mailbox_id={} account_id={}", - mailbox_id, account_id + "Deleted {} address entities for mailbox_id={} account_id={}", + deleted, mailbox_id, account_id ); Ok(()) } diff --git a/src/modules/cache/imap/manager.rs b/src/modules/cache/imap/manager.rs index e928f75..6154493 100644 --- a/src/modules/cache/imap/manager.rs +++ b/src/modules/cache/imap/manager.rs @@ -86,6 +86,10 @@ impl EnvelopeFlagsManager { mailbox_id: u64, to_delete_uid: &[u32], ) -> RustMailerResult<()> { + EmailEnvelopeV3::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?; + MinimalEnvelope::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?; + AddressEntity::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?; + EmailThread::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?; if let Some(mailboxes_map) = FLAGS_STATE_MAP.get(&account_id) { if let Some(flags_map) = mailboxes_map.get(&mailbox_id) { for uid in to_delete_uid { @@ -99,10 +103,7 @@ impl EnvelopeFlagsManager { FLAGS_STATE_MAP.remove(&account_id); } } - EmailEnvelopeV3::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?; - MinimalEnvelope::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?; - AddressEntity::clean_envelopes(account_id, mailbox_id, to_delete_uid).await?; - EmailThread::clean_envelopes(account_id, mailbox_id, to_delete_uid).await + Ok(()) } /// Clean all data associated with a specific mailbox for a given account. diff --git a/src/modules/cache/imap/migration.rs b/src/modules/cache/imap/migration.rs index 1ab46a8..7f73ea2 100644 --- a/src/modules/cache/imap/migration.rs +++ b/src/modules/cache/imap/migration.rs @@ -2,7 +2,7 @@ // Licensed under RustMailer License Agreement v1.0 // Unauthorized copying, modification, or distribution is prohibited. -use std::{collections::HashSet, sync::Arc}; +use std::sync::Arc; use native_db::*; use native_model::{native_model, Model}; @@ -26,9 +26,9 @@ use crate::{ }, common::Addr, database::{ - 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, + batch_delete_impl, 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}, @@ -475,27 +475,29 @@ impl EmailEnvelopeV3 { mailbox_id: u64, to_delete_uid: &[u32], ) -> RustMailerResult<()> { - const BATCH_SIZE: usize = 50; - let to_delete_set: HashSet = to_delete_uid.iter().copied().collect(); - let to_delete_set = Arc::new(to_delete_set); - let filter: RowFilter = 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 - ), - )?; + let hashes: Vec = to_delete_uid + .iter() + .map(|uid| envelope_hash(account_id, mailbox_id, *uid)) + .collect(); + + let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| { + let mut to_delete = Vec::with_capacity(hashes.len()); + for hash in hashes { + if let Some(envelope) = rw + .get() + .secondary::(EmailEnvelopeV3Key::create_envelope_id, hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + { + to_delete.push(envelope); + } + } + Ok(to_delete) + }) + .await?; info!( - "Enqueued deletion of envelopes for account_id={} mailbox_id={}", - account_id, mailbox_id + "Deleted {} envelopes for account_id={} mailbox_id={}", + deleted, account_id, mailbox_id ); Ok(()) } diff --git a/src/modules/cache/imap/minimal.rs b/src/modules/cache/imap/minimal.rs index 7bbddeb..03cc353 100644 --- a/src/modules/cache/imap/minimal.rs +++ b/src/modules/cache/imap/minimal.rs @@ -2,7 +2,7 @@ // Licensed under RustMailer License Agreement v1.0 // Unauthorized copying, modification, or distribution is prohibited. -use std::{collections::HashSet, sync::Arc}; +use std::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_insert_impl, enqueue_delete_secondary_impl, filter_by_secondary_key_impl, - manager::DB_MANAGER, safe_delete::RowFilter, update_impl, + batch_delete_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, @@ -95,27 +95,29 @@ impl MinimalEnvelope { mailbox_id: u64, to_delete_uid: &[u32], ) -> RustMailerResult<()> { - const BATCH_SIZE: usize = 50; - let to_delete_set: HashSet = to_delete_uid.iter().copied().collect(); - let to_delete_set = Arc::new(to_delete_set); - let filter: RowFilter = 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 - ), - )?; + let hashes: Vec = to_delete_uid + .iter() + .map(|uid| envelope_hash(account_id, mailbox_id, *uid)) + .collect(); + + let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| { + let mut to_delete = Vec::with_capacity(hashes.len()); + for hash in hashes { + if let Some(envelope) = rw + .get() + .primary::(hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + { + to_delete.push(envelope); + } + } + Ok(to_delete) + }) + .await?; info!( - "Enqueued deletion of minimal envelopes for account_id={} mailbox_id={}", - account_id, mailbox_id + "Deleted {} minimal envelopes for account_id={} mailbox_id={}", + deleted, account_id, mailbox_id ); Ok(()) } diff --git a/src/modules/cache/imap/thread.rs b/src/modules/cache/imap/thread.rs index 50b7f53..5a7db55 100644 --- a/src/modules/cache/imap/thread.rs +++ b/src/modules/cache/imap/thread.rs @@ -2,7 +2,7 @@ // Licensed under RustMailer License Agreement v1.0 // Unauthorized copying, modification, or distribution is prohibited. -use std::{collections::HashSet, sync::Arc}; +use std::sync::Arc; use futures::future::join_all; use native_db::*; @@ -23,8 +23,8 @@ use crate::{ }, }, database::{ - enqueue_delete_secondary_impl, manager::DB_MANAGER, paginate_secondary_scan_impl, - safe_delete::RowFilter, + batch_delete_impl, enqueue_delete_secondary_impl, manager::DB_MANAGER, + paginate_secondary_scan_impl, safe_delete::RowFilter, }, error::{code::ErrorCode, RustMailerResult}, rest::response::DataPage, @@ -112,32 +112,29 @@ impl EmailThread { mailbox_id: u64, to_delete_uid: &[u32], ) -> RustMailerResult<()> { - const BATCH_SIZE: usize = 50; - - let to_delete_set: HashSet = to_delete_uid + let hashes: Vec = to_delete_uid .iter() .map(|uid| envelope_hash(account_id, mailbox_id, *uid)) .collect(); - let to_delete_set = Arc::new(to_delete_set); - let filter: RowFilter = 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 - ), - )?; + let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| { + let mut to_delete = Vec::with_capacity(hashes.len()); + for hash in hashes { + if let Some(thread) = rw + .get() + .secondary::(EmailThreadKey::envelope_id, hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + { + to_delete.push(thread); + } + } + Ok(to_delete) + }) + .await?; info!( - "Enqueued deletion of thread entities for mailbox_id={} account_id={}", - mailbox_id, account_id + "Deleted {} thread entities for account_id={} mailbox_id={}", + deleted, account_id, mailbox_id ); Ok(()) } diff --git a/src/modules/database/safe_delete.rs b/src/modules/database/safe_delete.rs index 8e1d7d5..5fdc7a2 100644 --- a/src/modules/database/safe_delete.rs +++ b/src/modules/database/safe_delete.rs @@ -293,6 +293,7 @@ where #[cfg(test)] mod tests { use super::*; + use crate::modules::database::batch_delete_impl; use itertools::Itertools; use native_model::{native_model, Model}; use serde::{Deserialize, Serialize}; @@ -580,6 +581,163 @@ mod tests { } } + /// Production point-lookup delete (the stale-envelope cleanup fix): one bounded + /// write tx, one lookup per UID, no full mailbox scan. Each entity type must + /// remove exactly the target rows and leave everything else untouched. + #[tokio::test] + async fn point_lookup_delete_only_removes_target_rows() { + let db = new_bench_db(); + seed_minimal(&db); + seed_address(&db); + seed_thread(&db); + seed_envelope_v3(&db); + + let uids: Vec = vec![3, 17, 99]; + let addr_hash = |uid: u32| (uid as u64).wrapping_mul(0x9E3779B97F4A7C15); + + // One envelope can produce several AddressEntity rows (to/cc/...): all rows + // sharing the envelope_hash must go, not just one. + let extra = [ + BenchAddress { + id: 9_000_001, + account_id: BN_ACCT, + mailbox_id: BN_MB, + envelope_hash: addr_hash(3), + note: "extra-1".into(), + }, + BenchAddress { + id: 9_000_002, + account_id: BN_ACCT, + mailbox_id: BN_MB, + envelope_hash: addr_hash(3), + note: "extra-2".into(), + }, + ]; + let rw = db.rw_transaction().unwrap(); + for row in extra.iter().cloned() { + rw.insert(row).unwrap(); + } + rw.commit().unwrap(); + + // MinimalEnvelope: u64 primary key IS envelope_hash. + let hashes: Vec = uids + .iter() + .map(|uid| BenchMinimal { + account_id: BN_ACCT, + mailbox_id: BN_MB, + uid: *uid, + flags_hash: 0, + } + .pk()) + .collect(); + let deleted = batch_delete_impl(&db, move |rw| { + let mut to_delete = Vec::with_capacity(hashes.len()); + for hash in hashes { + if let Some(e) = rw + .get() + .primary::(hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + { + to_delete.push(e); + } + } + Ok(to_delete) + }) + .await + .unwrap(); + assert_eq!(deleted, uids.len()); + + // AddressEntity: non-unique envelope_hash point scan. + let hashes: Vec = uids.iter().map(|uid| addr_hash(*uid)).collect(); + let deleted = batch_delete_impl(&db, move |rw| { + let mut to_delete = Vec::new(); + for hash in hashes { + let rows: Vec = rw + .scan() + .secondary(BenchAddressKey::envelope_hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + .start_with(hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + .filter_map(Result::ok) + .collect(); + to_delete.extend(rows); + } + Ok(to_delete) + }) + .await + .unwrap(); + // 3 envelopes + 2 extra AddressEntity rows that share uid-3's hash. + assert_eq!(deleted, uids.len() + 2); + + // EmailThread: unique envelope_id secondary lookup. + let hashes: Vec = uids.iter().map(|uid| *uid as u64).collect(); + let deleted = batch_delete_impl(&db, move |rw| { + let mut to_delete = Vec::with_capacity(hashes.len()); + for hash in hashes { + if let Some(t) = rw + .get() + .secondary::(BenchThreadKey::envelope_id, hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + { + to_delete.push(t); + } + } + Ok(to_delete) + }) + .await + .unwrap(); + assert_eq!(deleted, uids.len()); + + // EmailEnvelopeV3: unique create_envelope_id secondary lookup. + let hashes: Vec = uids.iter().map(|uid| addr_hash(*uid)).collect(); + let deleted = batch_delete_impl(&db, move |rw| { + let mut to_delete = Vec::with_capacity(hashes.len()); + for hash in hashes { + if let Some(e) = rw + .get() + .secondary::(BenchEnvelopeV3Key::create_envelope_id, hash) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? + { + to_delete.push(e); + } + } + Ok(to_delete) + }) + .await + .unwrap(); + assert_eq!(deleted, uids.len()); + + // Only the target rows were removed; every other row survives. + let r = db.r_transaction().unwrap(); + let expect = BN - uids.len() as u64; + assert_eq!(r.len().primary::().unwrap(), expect); + assert_eq!(r.len().primary::().unwrap(), expect); + assert_eq!(r.len().primary::().unwrap(), expect); + assert_eq!(r.len().primary::().unwrap(), expect); + + let pk = |uid: u32| { + BenchMinimal { + account_id: BN_ACCT, + mailbox_id: BN_MB, + uid, + flags_hash: 0, + } + .pk() + }; + assert!(r.get().primary::(pk(1)).unwrap().is_some()); + assert!(r.get().primary::(pk(3)).unwrap().is_none()); + assert!(r + .get() + .secondary::(BenchEnvelopeV3Key::create_envelope_id, addr_hash(1)) + .unwrap() + .is_some()); + assert!(r + .get() + .secondary::(BenchEnvelopeV3Key::create_envelope_id, addr_hash(17)) + .unwrap() + .is_none()); + } + /// Failure mode of the current design: a scan that stalls inside the write tx /// blocks every other writer in the process. #[test]