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

Message caching improvements

Browse Source
This commit is contained in:
mdecimus
2025-05-06 16:00:16 +02:00
parent b8329afeba
commit e69b5ec2b7
87 changed files with 2835 additions and 2536 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
crates/common/src/config/inner.rs
View File

@@ -23,8 +23,8 @@ use utils::{
};
use crate::{
CacheSwap, Caches, Data, DavResource, DavResources, MailboxCache, MailboxStoreCache,
MessageStoreCache, MessageUidCache, TlsConnectors,
CacheSwap, Caches, Data, DavResource, DavResources, MailboxCache, MessageStoreCache,
MessageUidCache, TlsConnectors,
auth::{AccessToken, roles::RolePermissions},
config::smtp::resolver::{Policy, Tlsa},
listener::blocked::BlockedIps,
@@ -107,21 +107,14 @@ impl Caches {
MB_5,
std::mem::size_of::<RolePermissions>() as u64,
),
mailboxes: Cache::from_config(
config,
"mailbox",
MB_10,
(std::mem::size_of::<u32>()
+ std::mem::size_of::<CacheSwap<MailboxStoreCache>>()
+ (15 * (std::mem::size_of::<MailboxCache>() + 60))) as u64,
),
messages: Cache::from_config(
config,
"message",
MB_10,
(std::mem::size_of::<u32>()
+ std::mem::size_of::<CacheSwap<MessageStoreCache>>()
+ (1024 * std::mem::size_of::<MessageUidCache>())) as u64,
+ (1024 * std::mem::size_of::<MessageUidCache>())
+ (15 * (std::mem::size_of::<MailboxCache>() + 60))) as u64,
),
dav: Cache::from_config(
config,

6
crates/common/src/config/jmap/settings.rs
View File

@@ -16,7 +16,7 @@ pub struct JmapConfig {
pub query_max_results: usize,
pub snippet_max_results: usize,
pub changes_max_results: usize,
pub changes_max_results: Option<usize>,
pub changes_max_history: Option<Duration>,
pub request_max_size: usize,
@@ -242,8 +242,8 @@ impl JmapConfig {
.property("jmap.protocol.query.max-results")
.unwrap_or(5000),
changes_max_results: config
.property("jmap.protocol.changes.max-results")
.unwrap_or(5000),
.property_or_default::<Option<usize>>("jmap.protocol.changes.max-results", "5000")
.unwrap_or_default(),
changes_max_history: config
.property_or_default::<Option<Duration>>("jmap.protocol.changes.max-history", "30d")
.unwrap_or_default(),

37
crates/common/src/core.rs
View File

@@ -8,7 +8,10 @@ use std::{sync::Arc, time::Duration};
use directory::{Directory, QueryBy, Type, backend::internal::manage::ManageDirectory};
use jmap_proto::types::{
blob::BlobId, collection::Collection, property::Property, state::StateChange,
blob::BlobId,
collection::{Collection, SyncCollection},
property::Property,
state::StateChange,
type_state::DataType,
};
use sieve::Sieve;
@@ -506,10 +509,16 @@ impl Server {
}
if let Some(changes) = builder.changes() {
for (account_id, (change_id, changed_collections)) in changes {
for (account_id, changed_collections) in changes {
let mut state_change = StateChange::new(account_id);
for changed_collection in changed_collections {
if let Ok(data_type) = DataType::try_from(changed_collection) {
let change_id = changed_collections.change_id;
for changed_collection in changed_collections.changed_containers {
if let Some(data_type) = DataType::try_from_id(changed_collection, true) {
state_change.set_change(data_type, change_id);
}
}
for changed_collection in changed_collections.changed_items {
if let Some(data_type) = DataType::try_from_id(changed_collection, false) {
state_change.set_change(data_type, change_id);
}
}
@@ -531,18 +540,14 @@ impl Server {
})?;
for collection in [
Collection::Email.into(),
Collection::Mailbox.into(),
Collection::Mailbox.as_child_update(),
Collection::Thread.into(),
Collection::Identity.into(),
Collection::EmailSubmission.into(),
Collection::SieveScript.into(),
Collection::FileNode.into(),
Collection::AddressBook.into(),
Collection::ContactCard.into(),
Collection::Calendar.into(),
Collection::CalendarEvent.into(),
SyncCollection::Email.into(),
SyncCollection::Thread.into(),
SyncCollection::Identity.into(),
SyncCollection::EmailSubmission.into(),
SyncCollection::SieveScript.into(),
SyncCollection::FileNode.into(),
SyncCollection::AddressBook.into(),
SyncCollection::Calendar.into(),
] {
self.core
.storage

31
crates/common/src/lib.rs
View File

@@ -149,7 +149,6 @@ pub struct Caches {
pub permissions: Cache<u32, Arc<RolePermissions>>,
pub messages: Cache<u32, CacheSwap<MessageStoreCache>>,
pub mailboxes: Cache<u32, CacheSwap<MailboxStoreCache>>,
pub dav: Cache<DavResourceId, Arc<DavResources>>,
pub bayes: CacheWithTtl<TokenHash, Weights>,
@@ -168,21 +167,28 @@ pub struct Caches {
pub struct CacheSwap<T>(pub Arc<ArcSwap<T>>);
#[derive(Debug, Clone)]
pub struct MailboxStoreCache {
pub change_id: u64,
pub index: AHashMap<u32, u32>,
pub items: Vec<MailboxCache>,
pub struct MessageStoreCache {
pub emails: Arc<MessagesCache>,
pub mailboxes: Arc<MailboxesCache>,
pub update_lock: Arc<Semaphore>,
pub last_change_id: u64,
pub size: u64,
}
#[derive(Debug, Clone)]
pub struct MessageStoreCache {
pub struct MailboxesCache {
pub change_id: u64,
pub index: AHashMap<u32, u32>,
pub items: Vec<MailboxCache>,
pub size: u64,
}
#[derive(Debug, Clone)]
pub struct MessagesCache {
pub change_id: u64,
pub items: Vec<MessageCache>,
pub index: AHashMap<u32, u32>,
pub keywords: Vec<String>,
pub update_lock: Arc<Semaphore>,
pub size: u64,
}
@@ -311,12 +317,6 @@ impl CacheItemWeight for MessageStoreCache {
}
}
impl CacheItemWeight for MailboxStoreCache {
fn weight(&self) -> u64 {
self.size
}
}
impl CacheItemWeight for HttpAuthCache {
fn weight(&self) -> u64 {
std::mem::size_of::<HttpAuthCache>() as u64
@@ -427,7 +427,6 @@ impl Default for Caches {
access_tokens: Cache::new(1024, 10 * 1024 * 1024),
http_auth: Cache::new(1024, 10 * 1024 * 1024),
permissions: Cache::new(1024, 10 * 1024 * 1024),
mailboxes: Cache::new(1024, 10 * 1024 * 1024),
messages: Cache::new(1024, 25 * 1024 * 1024),
dav: Cache::new(1024, 10 * 1024 * 1024),
bayes: CacheWithTtl::new(1024, 10 * 1024 * 1024),
@@ -616,14 +615,14 @@ impl MessageStoreCache {
bytes.extend_from_slice(message_id);
let mut hash = store::gxhash::gxhash32(&bytes, 791120);
if self.items.is_empty() {
if self.emails.items.is_empty() {
return hash;
}
// Naive pass, assume hash is unique
let mut threads_ids = RoaringBitmap::new();
let mut is_unique_hash = true;
for item in self.items.iter() {
for item in self.emails.items.iter() {
if is_unique_hash && item.thread_id != hash {
is_unique_hash = false;
}

73
crates/common/src/storage/index.rs
View File

@@ -40,13 +40,17 @@ pub enum IndexValue<'x> {
Quota {
used: u32,
},
LogChild {
prefix: Option<u32>,
LogContainer {
sync_collection: u8,
},
LogParent {
collection: u8,
LogContainerProperty {
sync_collection: u8,
ids: Vec<u32>,
},
LogItem {
sync_collection: u8,
prefix: Option<u32>,
},
Acl {
value: Cow<'x, [AclGrant]>,
},
@@ -299,8 +303,19 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> IntoOperations
for (current, change) in current.inner.index_values().zip(changes.index_values()) {
if current != change {
merge_index(batch, current, change, self.tenant_id)?;
} else if let IndexValue::LogChild { prefix } = current {
batch.log_update(prefix);
} else {
match current {
IndexValue::LogContainer { sync_collection } => {
batch.log_container_update(sync_collection);
}
IndexValue::LogItem {
sync_collection,
prefix,
} => {
batch.log_item_update(sync_collection, prefix);
}
_ => (),
}
}
}
batch.set(Property::Value, Archiver::new(changes).serialize()?);
@@ -377,16 +392,29 @@ fn build_index(batch: &mut BatchBuilder, item: IndexValue<'_>, tenant_id: Option
batch.add(DirectoryClass::UsedQuota(tenant_id), value);
}
}
IndexValue::LogChild { prefix } => {
IndexValue::LogItem {
sync_collection,
prefix,
} => {
if set {
batch.log_insert(prefix);
batch.log_item_insert(sync_collection, prefix);
} else {
batch.log_delete(prefix);
batch.log_item_delete(sync_collection, prefix);
}
}
IndexValue::LogParent { collection, ids } => {
IndexValue::LogContainer { sync_collection } => {
if set {
batch.log_container_insert(sync_collection);
} else {
batch.log_container_delete(sync_collection);
}
}
IndexValue::LogContainerProperty {
sync_collection,
ids,
} => {
for parent_id in ids {
batch.log_parent_update(collection, parent_id);
batch.log_container_property_change(sync_collection, parent_id);
}
}
}
@@ -517,27 +545,32 @@ fn merge_index(
}
}
(
IndexValue::LogChild { prefix: old_prefix },
IndexValue::LogChild { prefix: new_prefix },
IndexValue::LogItem {
sync_collection,
prefix: old_prefix,
},
IndexValue::LogItem {
prefix: new_prefix, ..
},
) => {
batch.log_delete(old_prefix);
batch.log_insert(new_prefix);
batch.log_item_delete(sync_collection, old_prefix);
batch.log_item_insert(sync_collection, new_prefix);
}
(
IndexValue::LogParent {
collection,
IndexValue::LogContainerProperty {
sync_collection,
ids: old_ids,
},
IndexValue::LogParent { ids: new_ids, .. },
IndexValue::LogContainerProperty { ids: new_ids, .. },
) => {
for parent_id in &old_ids {
if !new_ids.contains(parent_id) {
batch.log_parent_update(collection, *parent_id);
batch.log_container_property_change(sync_collection, *parent_id);
}
}
for parent_id in new_ids {
if !old_ids.contains(&parent_id) {
batch.log_parent_update(collection, parent_id);
batch.log_container_property_change(sync_collection, parent_id);
}
}
}