filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Delete in-memory keys by prefix

Browse Source
This commit is contained in:
mdecimus
2024-12-26 19:01:49 +01:00
parent 1ba229aac7
commit e5ebc5ce33
13 changed files with 566 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
crates/jmap/src/api/management/principal.rs
View File

@@ -6,7 +6,7 @@
use std::sync::{atomic::Ordering, Arc};
use common::{auth::AccessToken, Server};
use common::{auth::AccessToken, Server, KV_BAYES_MODEL_USER};
use directory::{
backend::internal::{
lookup::DirectoryStore,
@@ -355,6 +355,23 @@ impl PrincipalManager for Server {
self.core.storage.fts.remove_all(account_id).await?;
}
// Delete bayes model
if self
.core
.spam
.bayes
.as_ref()
.map_or(false, |c| c.account_classify)
{
let mut key = Vec::with_capacity(std::mem::size_of::<u32>() + 1);
key.push(KV_BAYES_MODEL_USER);
key.extend_from_slice(&account_id.to_be_bytes());
if let Err(err) = self.in_memory_store().key_delete_prefix(&key).await {
trc::error!(err.details("Failed to delete user bayes model"));
}
}
// Remove entries from cache
self.inner
.data

61
crates/jmap/src/api/management/stores.rs
View File

@@ -9,7 +9,7 @@ use common::{
auth::AccessToken,
ipc::{HousekeeperEvent, PurgeType},
manager::webadmin::Resource,
Server,
*,
};
use directory::{
backend::internal::manage::{self, ManageDirectory},
@@ -108,7 +108,7 @@ impl ManageStore for Server {
// Validate the access token
access_token.assert_has_permission(Permission::PurgeDataStore)?;
let store = if let Some(id) = id {
let store = if let Some(id) = id.filter(|id| *id != "default") {
if let Some(store) = self.core.storage.stores.get(id) {
store.clone()
} else {
@@ -121,11 +121,11 @@ impl ManageStore for Server {
self.housekeeper_request(HousekeeperEvent::Purge(PurgeType::Data(store)))
.await
}
(Some("purge"), Some("lookup"), id, &Method::GET) => {
(Some("purge"), Some("in-memory"), id, &Method::GET) => {
// Validate the access token
access_token.assert_has_permission(Permission::PurgeInMemoryStore)?;
let store = if let Some(id) = id {
let store = if let Some(id) = id.filter(|id| *id != "default") {
if let Some(store) = self.core.storage.lookups.get(id) {
store.clone()
} else {
@@ -135,8 +135,57 @@ impl ManageStore for Server {
self.core.storage.lookup.clone()
};
self.housekeeper_request(HousekeeperEvent::Purge(PurgeType::Lookup(store)))
.await
let prefix = match path.get(4).copied() {
Some("acme") => vec![KV_ACME].into(),
Some("oauth") => vec![KV_OAUTH].into(),
Some("rate-rcpt") => vec![KV_RATE_LIMIT_RCPT].into(),
Some("rate-scan") => vec![KV_RATE_LIMIT_SCAN].into(),
Some("rate-loiter") => vec![KV_RATE_LIMIT_LOITER].into(),
Some("rate-auth") => vec![KV_RATE_LIMIT_AUTH].into(),
Some("rate-hash") => vec![KV_RATE_LIMIT_HASH].into(),
Some("rate-contact") => vec![KV_RATE_LIMIT_CONTACT].into(),
Some("rate-jmap") => vec![KV_RATE_LIMIT_JMAP].into(),
Some("rate-jmap-auth") => vec![KV_RATE_LIMIT_JMAP_AUTH].into(),
Some("rate-http-anonymous") => vec![KV_RATE_LIMIT_HTTP_ANONYM].into(),
Some("rate-imap") => vec![KV_RATE_LIMIT_IMAP].into(),
Some("reputation-ip") => vec![KV_REPUTATION_IP].into(),
Some("reputation-from") => vec![KV_REPUTATION_FROM].into(),
Some("reputation-domain") => vec![KV_REPUTATION_DOMAIN].into(),
Some("reputation-asn") => vec![KV_REPUTATION_ASN].into(),
Some("greylist") => vec![KV_GREYLIST].into(),
Some("bayes-account") => {
if let Some(account) = path.get(5).copied() {
let account_id = self
.core
.storage
.data
.get_principal_id(decode_path_element(account).as_ref())
.await?
.ok_or_else(|| trc::ManageEvent::NotFound.into_err())?;
let mut key = Vec::with_capacity(std::mem::size_of::<u32>() + 1);
key.push(KV_BAYES_MODEL_USER);
key.extend_from_slice(&account_id.to_be_bytes());
key.into()
} else {
vec![KV_BAYES_MODEL_USER].into()
}
}
Some("bayes-global") => vec![KV_BAYES_MODEL_GLOBAL].into(),
Some("trusted-reply") => vec![KV_TRUSTED_REPLY].into(),
Some("lock-purge-account") => vec![KV_LOCK_PURGE_ACCOUNT].into(),
Some("lock-queue-message") => vec![KV_LOCK_QUEUE_MESSAGE].into(),
Some("lock-queue-report") => vec![KV_LOCK_QUEUE_REPORT].into(),
Some("lock-email-task") => vec![KV_LOCK_EMAIL_TASK].into(),
Some("lock-housekeeper") => vec![KV_LOCK_HOUSEKEEPER].into(),
_ => None,
};
self.housekeeper_request(HousekeeperEvent::Purge(PurgeType::Lookup {
store,
prefix,
}))
.await
}
(Some("purge"), Some("account"), id, &Method::GET) => {
// Validate the access token

18
crates/jmap/src/services/housekeeper.rs
View File

@@ -375,7 +375,10 @@ pub fn spawn_housekeeper(inner: Arc<Inner>, mut rx: mpsc::Receiver<HousekeeperEv
PurgeType::Blobs { store, blob_store }
}
PurgeStore::Lookup(in_memory_store) => {
PurgeType::Lookup(in_memory_store)
PurgeType::Lookup {
store: in_memory_store,
prefix: None,
}
}
},
idx as u32,
@@ -638,7 +641,7 @@ impl Purge for Server {
.collect::<Vec<_>>()
.into(),
),
PurgeType::Lookup(_) => (
PurgeType::Lookup { prefix: None, .. } => (
"in-memory",
[2u8]
.into_iter()
@@ -646,6 +649,7 @@ impl Purge for Server {
.collect::<Vec<_>>()
.into(),
),
PurgeType::Lookup { .. } => ("in-memory-prefix", None),
PurgeType::Account(_) => ("account", None),
};
if let Some(lock_name) = &lock_name {
@@ -719,8 +723,14 @@ impl Purge for Server {
trc::error!(err.details("Failed to purge blob store"));
}
}
PurgeType::Lookup(store) => {
if let Err(err) = store.purge_in_memory_store().await {
PurgeType::Lookup { store, prefix } => {
if let Some(prefix) = prefix {
if let Err(err) = store.key_delete_prefix(&prefix).await {
trc::error!(err
.details("Failed to delete key prefix")
.ctx(trc::Key::Key, prefix));
}
} else if let Err(err) = store.purge_in_memory_store().await {
trc::error!(err.details("Failed to purge lookup store"));
}
}