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

Refactored local delivery to avoid mpsc channel

Browse Source
This commit is contained in:
mdecimus
2025-01-17 15:29:55 +01:00
parent 00ad5a5c44
commit 2eb388674d
117 changed files with 2611 additions and 2628 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/common/src/config/smtp/queue.rs
View File

@@ -85,7 +85,6 @@ pub struct QueueOutboundTimeout {
#[derive(Debug, Clone)]
pub struct QueueThrottle {
pub outbound_concurrency: usize,
pub local_concurrency: usize,
pub sender: Vec<Throttle>,
pub rcpt: Vec<Throttle>,
pub host: Vec<Throttle>,
@@ -204,7 +203,6 @@ impl Default for QueueConfig {
},
throttle: QueueThrottle {
outbound_concurrency: 25,
local_concurrency: 10,
sender: Default::default(),
rcpt: Default::default(),
host: Default::default(),
@@ -392,10 +390,6 @@ fn parse_queue_throttle(config: &mut Config) -> QueueThrottle {
.property_or_default::<usize>("queue.threads.remote", "25")
.unwrap_or(25)
.max(1),
local_concurrency: config
.property_or_default::<usize>("queue.threads.local", "10")
.unwrap_or(10)
.max(1),
};
let all_throttles = parse_throttle(

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

@@ -4,21 +4,33 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::sync::Arc;
use std::{sync::Arc, time::Duration};
use directory::{backend::internal::manage::ManageDirectory, Directory, Type};
use directory::{backend::internal::manage::ManageDirectory, Directory, QueryBy, Type};
use jmap_proto::types::{
blob::BlobId, collection::Collection, property::Property, state::StateChange,
};
use sieve::Sieve;
use store::{
write::{QueueClass, ValueClass},
BlobStore, FtsStore, InMemoryStore, IterateParams, Store, ValueKey,
dispatch::DocumentSet,
roaring::RoaringBitmap,
write::{
key::DeserializeBigEndian, log::ChangeLogBuilder, now, BatchBuilder, BitmapClass, BlobOp,
DirectoryClass, QueueClass, TagValue, ValueClass,
},
BitmapKey, BlobClass, BlobStore, Deserialize, FtsStore, InMemoryStore, IterateParams, LogKey,
Serialize, Store, ValueKey, U32_LEN,
};
use trc::AddContext;
use utils::BlobHash;
use crate::{
auth::{AccessToken, ResourceToken, TenantInfo},
config::smtp::{
auth::{ArcSealer, DkimSigner},
queue::RelayHost,
},
ipc::StateEvent,
ImapId, Inner, MailboxState, Server,
};
@@ -170,6 +182,249 @@ impl Server {
})
}
pub async fn get_used_quota(&self, account_id: u32) -> trc::Result<i64> {
self.core
.storage
.data
.get_counter(DirectoryClass::UsedQuota(account_id))
.await
.add_context(|err| err.caused_by(trc::location!()).account_id(account_id))
}
pub async fn has_available_quota(
&self,
quotas: &ResourceToken,
item_size: u64,
) -> trc::Result<()> {
if quotas.quota != 0 {
let used_quota = self.get_used_quota(quotas.account_id).await? as u64;
if used_quota + item_size > quotas.quota {
return Err(trc::LimitEvent::Quota
.into_err()
.ctx(trc::Key::Limit, quotas.quota)
.ctx(trc::Key::Size, used_quota));
}
}
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
if self.core.is_enterprise_edition() {
if let Some(tenant) = quotas.tenant.filter(|tenant| tenant.quota != 0) {
let used_quota = self.get_used_quota(tenant.id).await? as u64;
if used_quota + item_size > tenant.quota {
return Err(trc::LimitEvent::TenantQuota
.into_err()
.ctx(trc::Key::Limit, tenant.quota)
.ctx(trc::Key::Size, used_quota));
}
}
}
// SPDX-SnippetEnd
Ok(())
}
pub async fn get_resource_token(
&self,
access_token: &AccessToken,
account_id: u32,
) -> trc::Result<ResourceToken> {
Ok(if access_token.primary_id == account_id {
ResourceToken {
account_id,
quota: access_token.quota,
tenant: access_token.tenant,
}
} else {
let mut quotas = ResourceToken {
account_id,
..Default::default()
};
if let Some(principal) = self
.core
.storage
.directory
.query(QueryBy::Id(account_id), false)
.await
.add_context(|err| err.caused_by(trc::location!()).account_id(account_id))?
{
quotas.quota = principal.quota();
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
if self.core.is_enterprise_edition() {
if let Some(tenant_id) = principal.tenant() {
quotas.tenant = TenantInfo {
id: tenant_id,
quota: self
.core
.storage
.directory
.query(QueryBy::Id(tenant_id), false)
.await
.add_context(|err| {
err.caused_by(trc::location!()).account_id(tenant_id)
})?
.map(|tenant| tenant.quota())
.unwrap_or_default(),
}
.into();
}
}
// SPDX-SnippetEnd
}
quotas
})
}
pub async fn get_property<U>(
&self,
account_id: u32,
collection: Collection,
document_id: u32,
property: impl AsRef<Property> + Sync + Send,
) -> trc::Result<Option<U>>
where
U: Deserialize + 'static,
{
let property = property.as_ref();
self.core
.storage
.data
.get_value::<U>(ValueKey {
account_id,
collection: collection.into(),
document_id,
class: ValueClass::Property(property.into()),
})
.await
.add_context(|err| {
err.caused_by(trc::location!())
.account_id(account_id)
.collection(collection)
.document_id(document_id)
.id(property.to_string())
})
}
pub async fn get_properties<U, I, P>(
&self,
account_id: u32,
collection: Collection,
iterate: &I,
property: P,
) -> trc::Result<Vec<(u32, U)>>
where
I: DocumentSet + Send + Sync,
P: AsRef<Property> + Sync + Send,
U: Deserialize + 'static,
{
let property: u8 = property.as_ref().into();
let collection: u8 = collection.into();
let expected_results = iterate.len();
let mut results = Vec::with_capacity(expected_results);
self.core
.storage
.data
.iterate(
IterateParams::new(
ValueKey {
account_id,
collection,
document_id: iterate.min(),
class: ValueClass::Property(property),
},
ValueKey {
account_id,
collection,
document_id: iterate.max(),
class: ValueClass::Property(property),
},
),
|key, value| {
let document_id = key.deserialize_be_u32(key.len() - U32_LEN)?;
if iterate.contains(document_id) {
results.push((document_id, U::deserialize(value)?));
Ok(expected_results == 0 || results.len() < expected_results)
} else {
Ok(true)
}
},
)
.await
.add_context(|err| {
err.caused_by(trc::location!())
.account_id(account_id)
.collection(collection)
.id(property.to_string())
})
.map(|_| results)
}
pub async fn get_document_ids(
&self,
account_id: u32,
collection: Collection,
) -> trc::Result<Option<RoaringBitmap>> {
self.core
.storage
.data
.get_bitmap(BitmapKey::document_ids(account_id, collection))
.await
.add_context(|err| {
err.caused_by(trc::location!())
.account_id(account_id)
.collection(collection)
})
}
pub async fn get_tag(
&self,
account_id: u32,
collection: Collection,
property: impl AsRef<Property> + Sync + Send,
value: impl Into<TagValue<u32>> + Sync + Send,
) -> trc::Result<Option<RoaringBitmap>> {
let property = property.as_ref();
self.core
.storage
.data
.get_bitmap(BitmapKey {
account_id,
collection: collection.into(),
class: BitmapClass::Tag {
field: property.into(),
value: value.into(),
},
document_id: 0,
})
.await
.add_context(|err| {
err.caused_by(trc::location!())
.account_id(account_id)
.collection(collection)
.id(property.to_string())
})
}
pub fn notify_task_queue(&self) {
self.inner.ipc.index_tx.notify_one();
}
pub async fn total_queued_messages(&self) -> trc::Result<u64> {
let mut total = 0;
self.store()
@@ -190,6 +445,166 @@ impl Server {
.map(|_| total)
}
pub fn begin_changes(&self, account_id: u32) -> trc::Result<ChangeLogBuilder> {
self.assign_change_id(account_id)
.map(ChangeLogBuilder::with_change_id)
}
#[inline(always)]
pub fn assign_change_id(&self, _: u32) -> trc::Result<u64> {
self.generate_snowflake_id()
}
pub fn generate_snowflake_id(&self) -> trc::Result<u64> {
self.inner.data.jmap_id_gen.generate().ok_or_else(|| {
trc::StoreEvent::UnexpectedError
.into_err()
.caused_by(trc::location!())
.ctx(trc::Key::Reason, "Failed to generate snowflake id.")
})
}
pub async fn commit_changes(
&self,
account_id: u32,
mut changes: ChangeLogBuilder,
) -> trc::Result<u64> {
if changes.change_id == u64::MAX || changes.change_id == 0 {
changes.change_id = self.assign_change_id(account_id)?;
}
let state = changes.change_id;
let mut builder = BatchBuilder::new();
builder.with_account_id(account_id).custom(changes);
self.core
.storage
.data
.write(builder.build())
.await
.caused_by(trc::location!())
.map(|_| state)
}
pub async fn delete_changes(&self, account_id: u32, before: Duration) -> trc::Result<()> {
let reference_cid = self.inner.data.jmap_id_gen.past_id(before).ok_or_else(|| {
trc::StoreEvent::UnexpectedError
.caused_by(trc::location!())
.ctx(trc::Key::Reason, "Failed to generate reference change id.")
})?;
for collection in [
Collection::Email,
Collection::Mailbox,
Collection::Thread,
Collection::Identity,
Collection::EmailSubmission,
] {
self.core
.storage
.data
.delete_range(
LogKey {
account_id,
collection: collection.into(),
change_id: 0,
},
LogKey {
account_id,
collection: collection.into(),
change_id: reference_cid,
},
)
.await?;
}
Ok(())
}
pub async fn broadcast_state_change(&self, state_change: StateChange) -> bool {
match self
.inner
.ipc
.state_tx
.clone()
.send(StateEvent::Publish { state_change })
.await
{
Ok(_) => true,
Err(_) => {
trc::event!(
Server(trc::ServerEvent::ThreadError),
Details = "Error sending state change.",
CausedBy = trc::location!()
);
false
}
}
}
#[allow(clippy::blocks_in_conditions)]
pub async fn put_blob(
&self,
account_id: u32,
data: &[u8],
set_quota: bool,
) -> trc::Result<BlobId> {
// First reserve the hash
let hash = BlobHash::from(data);
let mut batch = BatchBuilder::new();
let until = now() + self.core.jmap.upload_tmp_ttl;
batch.with_account_id(account_id).set(
BlobOp::Reserve {
hash: hash.clone(),
until,
},
(if set_quota { data.len() as u32 } else { 0u32 }).serialize(),
);
self.core
.storage
.data
.write(batch.build())
.await
.caused_by(trc::location!())?;
if !self
.core
.storage
.data
.blob_exists(&hash)
.await
.caused_by(trc::location!())?
{
// Upload blob to store
self.core
.storage
.blob
.put_blob(hash.as_ref(), data)
.await
.caused_by(trc::location!())?;
// Commit blob
let mut batch = BatchBuilder::new();
batch.set(BlobOp::Commit { hash: hash.clone() }, Vec::new());
self.core
.storage
.data
.write(batch.build())
.await
.caused_by(trc::location!())?;
}
Ok(BlobId {
hash,
class: BlobClass::Reserved {
account_id,
expires: until,
},
section: None,
})
}
pub async fn total_accounts(&self) -> trc::Result<u64> {
self.store()
.count_principals(None, Type::Individual.into(), None)

36
crates/common/src/ipc.rs
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{borrow::Cow, sync::Arc, time::Instant};
use std::{sync::Arc, time::Instant};
use ahash::RandomState;
use jmap_proto::types::{state::StateChange, type_state::DataType};
@@ -14,8 +14,8 @@ use mail_auth::{
report::{tlsrpt::FailureDetails, Record},
};
use store::{BlobStore, InMemoryStore, Store};
use tokio::sync::{mpsc, oneshot};
use utils::{map::bitmap::Bitmap, BlobHash};
use tokio::sync::mpsc;
use utils::map::bitmap::Bitmap;
use crate::{
config::smtp::{
@@ -25,36 +25,6 @@ use crate::{
listener::limiter::ConcurrencyLimiter,
};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DeliveryResult {
Success,
TemporaryFailure {
reason: Cow<'static, str>,
},
PermanentFailure {
code: [u8; 3],
reason: Cow<'static, str>,
},
}
#[derive(Debug)]
pub enum DeliveryEvent {
Ingest {
message: IngestMessage,
result_tx: oneshot::Sender<Vec<DeliveryResult>>,
},
Stop,
}
#[derive(Debug)]
pub struct IngestMessage {
pub sender_address: String,
pub recipients: Vec<String>,
pub message_blob: BlobHash,
pub message_size: usize,
pub session_id: u64,
}
pub enum HousekeeperEvent {
AcmeReschedule {
provider_id: String,

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

@@ -33,7 +33,7 @@ use config::{
use dashmap::DashMap;
use imap_proto::protocol::list::Attribute;
use ipc::{DeliveryEvent, HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent};
use ipc::{HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent};
use listener::{
asn::AsnGeoLookupData, blocked::Security, limiter::ConcurrencyLimiter, tls::AcmeProviders,
};
@@ -43,7 +43,7 @@ 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::sync::{mpsc, Notify, Semaphore};
use tokio_rustls::TlsConnector;
use utils::{
cache::{Cache, CacheItemWeight, CacheWithTtl},
@@ -167,10 +167,10 @@ pub struct HttpAuthCache {
pub struct Ipc {
pub state_tx: mpsc::Sender<StateEvent>,
pub housekeeper_tx: mpsc::Sender<HousekeeperEvent>,
pub delivery_tx: mpsc::Sender<DeliveryEvent>,
pub index_tx: Arc<Notify>,
pub queue_tx: mpsc::Sender<QueueEvent>,
pub report_tx: mpsc::Sender<ReportingEvent>,
pub local_delivery_sm: Arc<Semaphore>,
}
pub struct TlsConnectors {
@@ -442,10 +442,10 @@ impl Default for Ipc {
Self {
state_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
housekeeper_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
delivery_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
index_tx: Default::default(),
queue_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
report_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
local_delivery_sm: Arc::new(Semaphore::new(10)),
}
}
}

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

@@ -16,7 +16,7 @@ use store::{
rand::{distributions::Alphanumeric, thread_rng, Rng},
Stores,
};
use tokio::sync::{mpsc, Notify};
use tokio::sync::{mpsc, Notify, Semaphore};
use utils::{
config::{Config, ConfigKey},
failed, Semver, UnwrapFailure,
@@ -25,7 +25,7 @@ use utils::{
use crate::{
config::{network::AsnGeoLookupConfig, server::Listeners, telemetry::Telemetry},
core::BuildServer,
ipc::{DeliveryEvent, HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent},
ipc::{HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent},
Caches, Core, Data, Inner, Ipc, IPC_CHANNEL_BUFFER,
};
@@ -46,7 +46,6 @@ pub struct BootManager {
pub struct IpcReceivers {
pub state_rx: Option<mpsc::Receiver<StateEvent>>,
pub housekeeper_rx: Option<mpsc::Receiver<HousekeeperEvent>>,
pub delivery_rx: Option<mpsc::Receiver<DeliveryEvent>>,
pub queue_rx: Option<mpsc::Receiver<QueueEvent>>,
pub report_rx: Option<mpsc::Receiver<ReportingEvent>>,
}
@@ -427,7 +426,7 @@ impl BootManager {
core.network.asn_geo_lookup,
AsnGeoLookupConfig::Resource { .. }
);
let (ipc, ipc_rxs) = build_ipc();
let (ipc, ipc_rxs) = build_ipc(&mut config);
let inner = Arc::new(Inner {
shared_core: ArcSwap::from_pointee(core),
data,
@@ -484,9 +483,8 @@ impl BootManager {
}
}
pub fn build_ipc() -> (Ipc, IpcReceivers) {
pub fn build_ipc(config: &mut Config) -> (Ipc, IpcReceivers) {
// Build ipc receivers
let (delivery_tx, delivery_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
let (state_tx, state_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
let (housekeeper_tx, housekeeper_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
let (queue_tx, queue_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
@@ -495,15 +493,19 @@ pub fn build_ipc() -> (Ipc, IpcReceivers) {
Ipc {
state_tx,
housekeeper_tx,
delivery_tx,
queue_tx,
report_tx,
index_tx: Arc::new(Notify::new()),
local_delivery_sm: Arc::new(Semaphore::new(
config
.property_or_default::<usize>("queue.threads.local", "10")
.unwrap_or(10)
.max(1),
)),
},
IpcReceivers {
state_rx: Some(state_rx),
housekeeper_rx: Some(housekeeper_rx),
delivery_rx: Some(delivery_rx),
queue_rx: Some(queue_rx),
report_rx: Some(report_rx),
},