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

S3-FIFO caching

Browse Source
This commit is contained in:
mdecimus
2024-12-27 19:40:33 +01:00
parent 3530b6625f
commit f2b00ccd54
33 changed files with 631 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
crates/common/src/auth/access_token.rs
View File

@@ -6,7 +6,7 @@
use directory::{
backend::internal::{lookup::DirectoryStore, PrincipalField},
Permission, Principal, QueryBy,
Permission, Principal, QueryBy, PERMISSIONS_BITSET_SIZE,
};
use jmap_proto::{
request::RequestMethod,
@@ -15,14 +15,15 @@ use jmap_proto::{
use std::{
hash::{DefaultHasher, Hash, Hasher},
sync::Arc,
time::Instant,
};
use store::query::acl::AclQuery;
use trc::AddContext;
use utils::map::{
bitmap::{Bitmap, BitmapItem},
ttl_dashmap::TtlMap,
vec_map::VecMap,
use utils::{
cache::TtlEntry,
map::{
bitmap::{Bitmap, BitmapItem},
vec_map::VecMap,
},
};
use crate::Server;
@@ -109,6 +110,7 @@ impl Server {
.unwrap_or_default(),
quota: principal.quota(),
permissions,
obj_size: 0,
})
}
@@ -183,27 +185,52 @@ impl Server {
}
}
Ok(access_token)
Ok(access_token.update_size())
}
pub fn cache_access_token(&self, access_token: Arc<AccessToken>) {
self.inner.data.access_tokens.insert_with_ttl(
access_token.primary_id(),
access_token,
Instant::now() + self.core.jmap.session_cache_ttl,
);
pub async fn get_or_build_access_token(
&self,
principal: Principal,
) -> trc::Result<Arc<AccessToken>> {
match self
.inner
.cache
.access_tokens
.get_value_or_guard_async(&principal.id())
.await
{
Ok(token) => Ok(token),
Err(guard) => {
let token = Arc::new(
self.update_access_token(self.build_access_token(principal).await?)
.await?,
);
let _ = guard.insert(TtlEntry::new(
token.clone(),
self.core.jmap.session_cache_ttl,
));
Ok(token)
}
}
}
pub async fn get_cached_access_token(&self, primary_id: u32) -> trc::Result<Arc<AccessToken>> {
if let Some(access_token) = self.inner.data.access_tokens.get_with_ttl(&primary_id) {
Ok(access_token)
} else {
// Refresh ACL token
self.get_access_token(primary_id).await.map(|access_token| {
let access_token = Arc::new(access_token);
self.cache_access_token(access_token.clone());
access_token
})
match self
.inner
.cache
.access_tokens
.get_value_or_guard_async(&primary_id)
.await
{
Ok(token) => Ok(token),
Err(guard) => {
let token = Arc::new(self.get_access_token(primary_id).await?);
let _ = guard.insert(TtlEntry::new(
token.clone(),
self.core.jmap.session_cache_ttl,
));
Ok(token)
}
}
}
}
@@ -467,4 +494,17 @@ impl AccessToken {
tenant: self.tenant,
}
}
pub fn update_size(mut self) -> Self {
self.obj_size = ((std::mem::size_of::<u32>() * 2)
+ (std::mem::size_of::<u64>() * 3)
+ (self.member_of.len() * std::mem::size_of::<u32>())
+ (self.access_to.len() * (std::mem::size_of::<u32>() + std::mem::size_of::<u64>()))
+ self.name.len()
+ self.description.as_ref().map_or(0, |v| v.len())
+ self.emails.iter().map(|v| v.len()).sum::<usize>()
+ (PERMISSIONS_BITSET_SIZE * std::mem::size_of::<usize>()))
as u64;
self
}
}

38
crates/common/src/auth/mod.rs
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{net::IpAddr, sync::Arc, time::Instant};
use std::{net::IpAddr, sync::Arc};
use directory::{
core::secret::verify_secret_hash, Directory, Permission, Permissions, Principal, QueryBy,
@@ -12,7 +12,10 @@ use directory::{
use jmap_proto::types::collection::Collection;
use mail_send::Credentials;
use oauth::GrantType;
use utils::map::{bitmap::Bitmap, ttl_dashmap::TtlMap, vec_map::VecMap};
use utils::{
cache::CacheItemWeight,
map::{bitmap::Bitmap, vec_map::VecMap},
};
use crate::Server;
@@ -32,6 +35,7 @@ pub struct AccessToken {
pub quota: u64,
pub permissions: Permissions,
pub tenant: Option<TenantInfo>,
pub obj_size: u64,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -72,21 +76,7 @@ impl Server {
}
}
_ => match self.authenticate_credentials(req, directory).await {
Ok(principal) => {
if let Some(access_token) =
self.inner.data.access_tokens.get_with_ttl(&principal.id())
{
Ok(access_token)
} else {
self.build_access_token(principal)
.await
.map(|access_token| {
let access_token = Arc::new(access_token);
self.cache_access_token(access_token.clone());
access_token
})
}
}
Ok(principal) => self.get_or_build_access_token(principal).await,
Err(err) => Err(err),
},
}
@@ -195,14 +185,6 @@ impl Server {
))
}
}
pub fn cache_session(&self, session_id: String, access_token: &AccessToken) {
self.inner.data.http_auth_cache.insert_with_ttl(
session_id,
access_token.primary_id(),
Instant::now() + self.core.jmap.session_cache_ttl,
);
}
}
impl<'x> AuthRequest<'x> {
@@ -247,6 +229,12 @@ impl<'x> AuthRequest<'x> {
}
}
impl CacheItemWeight for AccessToken {
fn weight(&self) -> u64 {
self.obj_size
}
}
pub(crate) trait CredentialsUsername {
fn login(&self) -> Option<&str>;
}

39
crates/common/src/auth/roles.rs
View File

@@ -9,9 +9,11 @@ use std::sync::{Arc, LazyLock};
use ahash::AHashSet;
use directory::{
backend::internal::{lookup::DirectoryStore, PrincipalField},
Permission, Permissions, QueryBy, ROLE_ADMIN, ROLE_TENANT_ADMIN, ROLE_USER,
Permission, Permissions, QueryBy, PERMISSIONS_BITSET_SIZE, ROLE_ADMIN, ROLE_TENANT_ADMIN,
ROLE_USER,
};
use trc::AddContext;
use utils::cache::CacheItemWeight;
use crate::Server;
@@ -33,10 +35,19 @@ impl Server {
ROLE_ADMIN => Ok(ADMIN_PERMISSIONS.clone()),
ROLE_TENANT_ADMIN => Ok(TENANT_ADMIN_PERMISSIONS.clone()),
role_id => {
if let Some(role_permissions) = self.inner.data.permissions.get(&role_id) {
Ok(role_permissions.clone())
} else {
self.build_role_permissions(role_id).await
match self
.inner
.cache
.permissions
.get_value_or_guard_async(&role_id)
.await
{
Ok(permissions) => Ok(permissions),
Err(guard) => {
let permissions = self.build_role_permissions(role_id).await?;
let _ = guard.insert(permissions.clone());
Ok(permissions)
}
}
}
}
@@ -81,7 +92,7 @@ impl Server {
}
role_id => {
// Try with the cache
if let Some(role_permissions) = self.inner.data.permissions.get(&role_id) {
if let Some(role_permissions) = self.inner.cache.permissions.get(&role_id) {
return_permissions.union(role_permissions.as_ref());
} else {
let mut role_permissions = RolePermissions::default();
@@ -133,7 +144,7 @@ impl Server {
} else {
// Cache role
self.inner
.data
.cache
.permissions
.insert(role_id, Arc::new(role_permissions));
}
@@ -147,13 +158,7 @@ impl Server {
}
}
// Cache role
let return_permissions = Arc::new(return_permissions);
self.inner
.data
.permissions
.insert(role_id, return_permissions.clone());
Ok(return_permissions)
Ok(Arc::new(return_permissions))
}
}
@@ -213,3 +218,9 @@ fn admin_permissions() -> Arc<RolePermissions> {
disabled: Permissions::new(),
})
}
impl CacheItemWeight for RolePermissions {
fn weight(&self) -> u64 {
(PERMISSIONS_BITSET_SIZE * std::mem::size_of::<usize>() * 2) as u64
}
}

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

@@ -12,14 +12,13 @@ use dashmap::DashMap;
use mail_send::smtp::tls::build_tls_connector;
use parking_lot::RwLock;
use utils::{
cache::{Cache, CacheWithTtl},
config::Config,
lru_cache::{LruCache, LruCached},
map::ttl_dashmap::{TtlDashMap, TtlMap},
snowflake::SnowflakeIdGenerator,
};
use crate::{
listener::blocked::BlockedIps, manager::webadmin::WebAdminManager, Data,
listener::blocked::BlockedIps, manager::webadmin::WebAdminManager, Caches, Data,
ThrottleKeyHasherBuilder, TlsConnectors,
};
@@ -59,12 +58,8 @@ impl Data {
})
.ok()
.map(Arc::new),
access_tokens: TtlDashMap::with_capacity(capacity, shard_amount),
http_auth_cache: TtlDashMap::with_capacity(capacity, shard_amount),
blocked_ips: RwLock::new(BlockedIps::parse(config).blocked_ip_addresses),
blocked_ips_version: 0.into(),
permissions: Default::default(),
permissions_version: 0.into(),
jmap_id_gen: id_generator.clone(),
queue_id_gen: id_generator.clone(),
span_id_gen: id_generator,
@@ -83,15 +78,6 @@ impl Data {
RandomState::default(),
shard_amount,
),
account_cache: LruCache::with_capacity(
config.property("cache.account.size").unwrap_or(2048),
),
mailbox_cache: LruCache::with_capacity(
config.property("cache.mailbox.size").unwrap_or(2048),
),
threads_cache: LruCache::with_capacity(
config.property("cache.thread.size").unwrap_or(2048),
),
logos: Default::default(),
smtp_session_throttle: DashMap::with_capacity_and_hasher_and_shard_amount(
capacity,
@@ -109,17 +95,29 @@ impl Data {
}
}
impl Caches {
pub fn parse(config: &mut Config) -> Self {
Caches {
access_tokens: CacheWithTtl::from_config(config, "cache.access-tokens"),
http_auth: CacheWithTtl::from_config(config, "cache.http-auth"),
permissions: Cache::from_config(config, "cache.permissions"),
permissions_version: 0.into(),
account: Cache::from_config(config, "cache.account"),
mailbox: Cache::from_config(config, "cache.mailbox"),
threads: Cache::from_config(config, "cache.threads"),
bayes: CacheWithTtl::from_config(config, "cache.bayes"),
dnsbl: CacheWithTtl::from_config(config, "cache.dnsbl"),
}
}
}
impl Default for Data {
fn default() -> Self {
Self {
tls_certificates: Default::default(),
tls_self_signed_cert: Default::default(),
access_tokens: Default::default(),
http_auth_cache: Default::default(),
blocked_ips: Default::default(),
blocked_ips_version: 0.into(),
permissions: Default::default(),
permissions_version: 0.into(),
jmap_id_gen: Default::default(),
queue_id_gen: Default::default(),
span_id_gen: Default::default(),
@@ -127,9 +125,6 @@ impl Default for Data {
config_version: Default::default(),
jmap_limiter: Default::default(),
imap_limiter: Default::default(),
account_cache: LruCache::with_capacity(2048),
mailbox_cache: LruCache::with_capacity(2048),
threads_cache: LruCache::with_capacity(2048),
logos: Default::default(),
smtp_session_throttle: Default::default(),
smtp_queue_throttle: Default::default(),

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

@@ -27,13 +27,13 @@ use listener::{
};
use manager::webadmin::{Resource, WebAdminManager};
use nlp::bayes::{TokenHash, Weights};
use parking_lot::{Mutex, RwLock};
use rustls::sign::CertifiedKey;
use tokio::sync::{mpsc, Notify};
use tokio_rustls::TlsConnector;
use utils::{
lru_cache::LruCache,
map::ttl_dashmap::{ADashMap, TtlDashMap},
cache::{Cache, CacheItemWeight, CacheWithTtl},
snowflake::SnowflakeIdGenerator,
};
@@ -92,6 +92,7 @@ pub struct Server {
pub struct Inner {
pub shared_core: ArcSwap<Core>,
pub data: Data,
pub cache: Caches,
pub ipc: Ipc,
}
@@ -99,15 +100,9 @@ pub struct Data {
pub tls_certificates: ArcSwap<AHashMap<String, Arc<CertifiedKey>>>,
pub tls_self_signed_cert: Option<Arc<CertifiedKey>>,
pub access_tokens: TtlDashMap<u32, Arc<AccessToken>>,
pub http_auth_cache: TtlDashMap<String, u32>,
pub blocked_ips: RwLock<AHashSet<IpAddr>>,
pub blocked_ips_version: AtomicU8,
pub permissions: ADashMap<u32, Arc<RolePermissions>>,
pub permissions_version: AtomicU8,
pub asn_geo_data: AsnGeoLookupData,
pub jmap_id_gen: SnowflakeIdGenerator,
@@ -120,10 +115,6 @@ pub struct Data {
pub jmap_limiter: DashMap<u32, Arc<ConcurrencyLimiters>, RandomState>,
pub imap_limiter: DashMap<u32, Arc<ConcurrencyLimiters>, RandomState>,
pub account_cache: LruCache<AccountId, Arc<Account>>,
pub mailbox_cache: LruCache<MailboxId, Arc<MailboxState>>,
pub threads_cache: LruCache<u32, Arc<Threads>>,
pub logos: Mutex<AHashMap<String, Option<Resource<Vec<u8>>>>>,
pub smtp_session_throttle: DashMap<ThrottleKey, ConcurrencyLimiter, ThrottleKeyHasherBuilder>,
@@ -131,6 +122,21 @@ pub struct Data {
pub smtp_connectors: TlsConnectors,
}
pub struct Caches {
pub access_tokens: CacheWithTtl<u32, Arc<AccessToken>>,
pub http_auth: CacheWithTtl<String, u32>,
pub permissions: Cache<u32, Arc<RolePermissions>>,
pub permissions_version: AtomicU8,
pub account: Cache<AccountId, Arc<Account>>,
pub mailbox: Cache<MailboxId, Arc<MailboxState>>,
pub threads: Cache<u32, Arc<Threads>>,
pub bayes: CacheWithTtl<TokenHash, Option<Weights>>,
pub dnsbl: CacheWithTtl<TokenHash, Option<Arc<Vec<IpAddr>>>>,
}
pub struct Ipc {
pub state_tx: mpsc::Sender<StateEvent>,
pub housekeeper_tx: mpsc::Sender<HousekeeperEvent>,
@@ -165,6 +171,7 @@ pub struct Account {
pub mailbox_state: AHashMap<u32, Mailbox>,
pub state_email: Option<u64>,
pub state_mailbox: Option<u64>,
pub obj_size: u64,
}
#[derive(Debug, Default, Clone)]
@@ -190,6 +197,7 @@ pub struct MailboxState {
pub total_messages: usize,
pub modseq: Option<u64>,
pub next_state: Option<Box<NextMailboxState>>,
pub obj_size: u64,
}
#[derive(Debug, Clone)]
@@ -231,6 +239,49 @@ pub struct Core {
pub enterprise: Option<enterprise::Enterprise>,
}
impl CacheItemWeight for AccountId {
fn weight(&self) -> u64 {
(std::mem::size_of::<u32>() * 2) as u64
}
}
impl CacheItemWeight for MailboxId {
fn weight(&self) -> u64 {
(std::mem::size_of::<u32>() * 2) as u64
}
}
impl CacheItemWeight for Threads {
fn weight(&self) -> u64 {
((self.threads.len() + 1) * std::mem::size_of::<u64>()) as u64
}
}
impl CacheItemWeight for MailboxState {
fn weight(&self) -> u64 {
self.obj_size
}
}
impl CacheItemWeight for Account {
fn weight(&self) -> u64 {
self.obj_size
}
}
impl MailboxState {
pub fn calculate_weight(&self) -> u64 {
(std::mem::size_of::<u64>() * 5) as u64
+ (self.id_to_imap.len() * std::mem::size_of::<u64>() + std::mem::size_of::<u32>())
as u64
+ (self.uid_to_id.len() * std::mem::size_of::<u64>()) as u64
+ self
.next_state
.as_ref()
.map_or(0, |n| n.next_state.calculate_weight())
}
}
pub trait IntoString: Sized {
fn into_string(self) -> String;
}
@@ -322,6 +373,25 @@ impl Default for Inner {
shared_core: Default::default(),
data: Default::default(),
ipc: Default::default(),
cache: Default::default(),
}
}
}
#[cfg(feature = "test_mode")]
#[allow(clippy::derivable_impls)]
impl Default for Caches {
fn default() -> Self {
Self {
access_tokens: CacheWithTtl::new(1024, 10 * 1024 * 1024),
http_auth: CacheWithTtl::new(1024, 10 * 1024 * 1024),
permissions: Cache::new(1024, 10 * 1024 * 1024),
permissions_version: Default::default(),
account: Cache::new(1024, 10 * 1024 * 1024),
mailbox: Cache::new(1024, 10 * 1024 * 1024),
threads: Cache::new(1024, 10 * 1024 * 1024),
bayes: CacheWithTtl::new(1024, 10 * 1024 * 1024),
dnsbl: CacheWithTtl::new(1024, 10 * 1024 * 1024),
}
}
}

6
crates/common/src/manager/boot.rs
View File

@@ -21,7 +21,7 @@ use utils::{
use crate::{
config::{server::Listeners, telemetry::Telemetry},
ipc::{DeliveryEvent, HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent},
Core, Data, Inner, Ipc, IPC_CHANNEL_BUFFER,
Caches, Core, Data, Inner, Ipc, IPC_CHANNEL_BUFFER,
};
use super::{
@@ -334,6 +334,9 @@ impl BootManager {
// Parse data
let data = Data::parse(&mut config);
// Parse caches
let cache = Caches::parse(&mut config);
// Enable telemetry
#[cfg(feature = "enterprise")]
telemetry.enable(core.is_enterprise_edition());
@@ -365,6 +368,7 @@ impl BootManager {
shared_core: ArcSwap::from_pointee(core),
data,
ipc,
cache,
});
// Parse TCP acceptors