fix(cache): use point query instead of scan when deleting local cache

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<u64> = to_delete_uid
+        let hashes: Vec<u64> = 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<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
-            ),
-        )?;
+        let deleted = batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
+            let mut to_delete = Vec::new();
+            for hash in hashes {
+                let entities: Vec<AddressEntity> = 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(())
     }

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.

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<u32> = to_delete_uid.iter().copied().collect();
-        let to_delete_set = Arc::new(to_delete_set);
-        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
-            ),
-        )?;
+        let hashes: Vec<u64> = 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::<EmailEnvelopeV3>(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(())
     }

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<u32> = to_delete_uid.iter().copied().collect();
-        let to_delete_set = Arc::new(to_delete_set);
-        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
-            ),
-        )?;
+        let hashes: Vec<u64> = 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::<MinimalEnvelope>(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(())
     }

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<u64> = to_delete_uid
+        let hashes: Vec<u64> = 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<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
-            ),
-        )?;
+        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::<EmailThread>(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(())
     }

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<u32> = 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<u64> = 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::<BenchMinimal>(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<u64> = 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<BenchAddress> = 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<u64> = 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::<BenchThread>(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<u64> = 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::<BenchEnvelopeV3>(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::<BenchMinimal>().unwrap(), expect);
+        assert_eq!(r.len().primary::<BenchAddress>().unwrap(), expect);
+        assert_eq!(r.len().primary::<BenchThread>().unwrap(), expect);
+        assert_eq!(r.len().primary::<BenchEnvelopeV3>().unwrap(), expect);
+
+        let pk = |uid: u32| {
+            BenchMinimal {
+                account_id: BN_ACCT,
+                mailbox_id: BN_MB,
+                uid,
+                flags_hash: 0,
+            }
+            .pk()
+        };
+        assert!(r.get().primary::<BenchMinimal>(pk(1)).unwrap().is_some());
+        assert!(r.get().primary::<BenchMinimal>(pk(3)).unwrap().is_none());
+        assert!(r
+            .get()
+            .secondary::<BenchEnvelopeV3>(BenchEnvelopeV3Key::create_envelope_id, addr_hash(1))
+            .unwrap()
+            .is_some());
+        assert!(r
+            .get()
+            .secondary::<BenchEnvelopeV3>(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]