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

JMAP Registry API implementation - part 9

Browse Source
This commit is contained in:
mdecimus
2026-03-05 19:07:35 +01:00
parent 92c60fff7a
commit 9f9ddee965
47 changed files with 3354 additions and 2280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
crates/common/src/cache/reload.rs vendored
View File

@@ -11,7 +11,7 @@ use crate::{
storage::Storage,
telemetry::Telemetry,
},
ipc::RegistryChange,
ipc::{QueueEvent, RegistryChange},
};
use ahash::AHashMap;
use directory::Directories;
@@ -142,4 +142,16 @@ impl Server {
Ok(result)
}
pub async fn reload_core(&self, new_core: Core) {
self.inner.shared_core.store(new_core.into());
// Reload queue settings
self.inner
.ipc
.queue_tx
.send(QueueEvent::ReloadSettings)
.await
.ok();
}
}

67
crates/common/src/config/network.rs
View File

@@ -57,18 +57,20 @@ pub struct ContactForm {
#[derive(Clone, Default)]
pub struct ClusterRoles {
pub purge_stores: ClusterRole,
pub purge_accounts: ClusterRole,
pub store_maintenance: ClusterRole,
pub account_maintenance: ClusterRole,
pub push_notifications: ClusterRole,
pub fts_indexing: ClusterRole,
pub search_indexing: ClusterRole,
pub spam_training: ClusterRole,
pub imip_processing: ClusterRole,
pub merge_threads: ClusterRole,
pub calendar_alerts: ClusterRole,
pub renew_acme: ClusterRole,
pub dns_acme: ClusterRole,
pub calculate_metrics: ClusterRole,
pub push_metrics: ClusterRole,
pub outbound_mta: ClusterRole,
pub task_scheduler: ClusterRole,
pub task_manager: ClusterRole,
}
#[derive(Clone, Copy, Default)]
@@ -171,40 +173,46 @@ impl Network {
let is_success = match &range.object {
NodeRole::CalculateMetrics(_)
| NodeRole::PushMetrics(_)
| NodeRole::TrainSpamClassifier(_) => {
| NodeRole::SpamClassifierTraining(_)
| NodeRole::TaskScheduler(_) => {
let (roles, role_obj) = match &range.object {
NodeRole::CalculateMetrics(role) => {
(&mut network.roles.calculate_metrics, role)
}
NodeRole::PushMetrics(role) => (&mut network.roles.push_metrics, role),
NodeRole::TrainSpamClassifier(role) => {
NodeRole::SpamClassifierTraining(role) => {
(&mut network.roles.spam_training, role)
}
NodeRole::TaskScheduler(role) => {
(&mut network.roles.task_scheduler, role)
}
_ => unreachable!(),
};
roles.set_role(role_obj.node_id == node_id)
}
NodeRole::PurgeStores(_)
| NodeRole::PurgeAccounts(_)
| NodeRole::AcmeRenew(_)
NodeRole::StoreMaintenance(_)
| NodeRole::AccountMaintenance(_)
| NodeRole::PushNotifications(_)
| NodeRole::SearchIndexing(_)
| NodeRole::ImipProcessing(_)
| NodeRole::CalendarAlerts(_)
| NodeRole::MergeThreads(_)
| NodeRole::OutboundMta(_) => {
| NodeRole::DnsAndAcme(_)
| NodeRole::OutboundMta(_)
| NodeRole::TaskQueueProcessing(_) => {
let (roles, role_obj) = match &range.object {
NodeRole::PurgeStores(role) => (&mut network.roles.purge_stores, role),
NodeRole::PurgeAccounts(role) => {
(&mut network.roles.purge_accounts, role)
NodeRole::StoreMaintenance(role) => {
(&mut network.roles.store_maintenance, role)
}
NodeRole::AccountMaintenance(role) => {
(&mut network.roles.account_maintenance, role)
}
NodeRole::AcmeRenew(role) => (&mut network.roles.renew_acme, role),
NodeRole::PushNotifications(role) => {
(&mut network.roles.push_notifications, role)
}
NodeRole::SearchIndexing(role) => {
(&mut network.roles.fts_indexing, role)
(&mut network.roles.search_indexing, role)
}
NodeRole::ImipProcessing(role) => {
(&mut network.roles.imip_processing, role)
@@ -216,6 +224,10 @@ impl Network {
(&mut network.roles.merge_threads, role)
}
NodeRole::OutboundMta(role) => (&mut network.roles.outbound_mta, role),
NodeRole::DnsAndAcme(role) => (&mut network.roles.dns_acme, role),
NodeRole::TaskQueueProcessing(role) => {
(&mut network.roles.task_manager, role)
}
_ => unreachable!(),
};
@@ -258,11 +270,11 @@ impl Network {
}
let roles = match shard_type {
NodeShardType::PurgeStores => &mut network.roles.purge_stores,
NodeShardType::PurgeAccounts => &mut network.roles.purge_accounts,
NodeShardType::AcmeRenew => &mut network.roles.renew_acme,
NodeShardType::StoreMaintenance => &mut network.roles.store_maintenance,
NodeShardType::AccountMaintenance => &mut network.roles.account_maintenance,
NodeShardType::DnsAndAcme => &mut network.roles.dns_acme,
NodeShardType::PushNotifications => &mut network.roles.push_notifications,
NodeShardType::SearchIndexing => &mut network.roles.fts_indexing,
NodeShardType::SearchIndexing => &mut network.roles.search_indexing,
NodeShardType::ImipProcessing => &mut network.roles.imip_processing,
NodeShardType::CalendarAlerts => &mut network.roles.calendar_alerts,
NodeShardType::MergeThreads => &mut network.roles.merge_threads,
@@ -419,7 +431,7 @@ impl ClusterRole {
matches!(self, ClusterRole::Enabled | ClusterRole::Sharded { .. })
}
pub fn is_enabled_for_integer(&self, value: u32) -> bool {
pub fn is_enabled_for_integer(&self, value: u64) -> bool {
debug_assert!(!self.is_uninit() && !self.is_seen_role());
match self {
ClusterRole::Enabled => true,
@@ -427,7 +439,7 @@ impl ClusterRole {
ClusterRole::Sharded {
shard_id,
total_shards,
} => (value % total_shards) == *shard_id,
} => (value as u32 % total_shards) == *shard_id,
}
}
@@ -440,7 +452,7 @@ impl ClusterRole {
shard_id,
total_shards,
} => {
let mut hasher = Xxh3Builder::new().with_seed(191179).build();
let mut hasher = Xxh3Builder::new().with_seed(201179).build();
item.hash(&mut hasher);
hasher.finish() % (*total_shards as u64) == *shard_id as u64
}
@@ -496,17 +508,20 @@ impl ClusterRole {
impl ClusterRoles {
fn all_mut(&mut self) -> impl Iterator<Item = &mut ClusterRole> {
[
&mut self.purge_stores,
&mut self.purge_accounts,
&mut self.store_maintenance,
&mut self.account_maintenance,
&mut self.push_notifications,
&mut self.fts_indexing,
&mut self.search_indexing,
&mut self.spam_training,
&mut self.imip_processing,
&mut self.merge_threads,
&mut self.calendar_alerts,
&mut self.renew_acme,
&mut self.dns_acme,
&mut self.outbound_mta,
&mut self.calculate_metrics,
&mut self.push_metrics,
&mut self.task_manager,
&mut self.task_scheduler,
]
.into_iter()
}

17
crates/common/src/enterprise/alerts.rs
View File

@@ -60,6 +60,17 @@ impl Server {
subject,
body,
} => {
let subject = subject.build();
trc::event!(
Telemetry(TelemetryEvent::AlertMessage),
Id = alert.id.id().id(),
To = to
.iter()
.map(|t| trc::Value::from(t.to_string()))
.collect::<Vec<_>>(),
Details = subject.clone()
);
messages.push(AlertMessage {
from: from_addr.clone(),
to: to.clone(),
@@ -82,7 +93,7 @@ impl Server {
)),
)
.header("Auto-Submitted", HeaderType::Text("auto-generated".into()))
.subject(subject.build())
.subject(subject)
.text_body(body.build())
.write_to_vec()
.unwrap_or_default(),
@@ -90,14 +101,14 @@ impl Server {
}
AlertMethod::Event { message } => {
trc::event!(
Telemetry(TelemetryEvent::Alert),
Telemetry(TelemetryEvent::AlertEvent),
Id = alert.id.id().id(),
Details = message.as_ref().map(|m| m.build())
);
#[cfg(feature = "test_mode")]
Collector::update_event_counter(
trc::EventType::Telemetry(TelemetryEvent::Alert),
trc::EventType::Telemetry(TelemetryEvent::AlertEvent),
1,
);
}

5
crates/common/src/enterprise/config.rs
View File

@@ -170,9 +170,12 @@ impl Enterprise {
// Build the enterprise configuration
let mut enterprise = Enterprise {
license,
undelete_retention: dr
deleted_items_retention: dr
.archive_deleted_items_for
.map(|retention| retention.into_inner()),
deleted_accounts_retention: dr
.archive_deleted_accounts_for
.map(|retention| retention.into_inner()),
logo_url,
metrics_alerts: Default::default(),
spam_filter_llm: SpamFilterLlmConfig::parse(bp, &ai_apis_ids).await,

3
crates/common/src/enterprise/mod.rs
View File

@@ -33,7 +33,8 @@ use utils::{HttpLimitResponse, cron::SimpleCron, template::Template};
pub struct Enterprise {
pub license: LicenseKey,
pub logo_url: Option<String>,
pub undelete_retention: Option<Duration>,
pub deleted_items_retention: Option<Duration>,
pub deleted_accounts_retention: Option<Duration>,
pub trace_retention: Option<Duration>,
pub metrics_retention: Option<Duration>,
pub metrics_interval: SimpleCron,

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

@@ -16,39 +16,14 @@ use mail_auth::{
report::{Record, tlsrpt::FailureDetails},
};
use registry::{schema::prelude::ObjectType, types::id::ObjectId};
use std::{
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
time::Instant,
use std::sync::{
Arc,
atomic::{AtomicBool, Ordering},
};
use tokio::sync::{Semaphore, SemaphorePermit, mpsc};
use types::type_state::{DataType, StateChange};
use utils::map::bitmap::Bitmap;
pub enum HousekeeperEvent {
AcmeReschedule {
provider_id: String,
renew_at: Instant,
},
Purge(PurgeType),
ReloadSettings,
Exit,
}
pub enum PurgeType {
Data,
Blob,
Lookup {
prefix: Option<Vec<u8>>,
},
Account {
account_id: Option<u32>,
use_roles: bool,
},
}
#[derive(Debug)]
pub enum PushEvent {
Subscribe {

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

@@ -38,7 +38,7 @@ use config::{
storage::Storage,
telemetry::Metrics,
};
use ipc::{BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent};
use ipc::{BroadcastEvent, PushEvent, QueueEvent, ReportingEvent};
use mail_auth::{MX, Txt};
use manager::application::{Resource, WebApplicationManager};
use parking_lot::{Mutex, RwLock};
@@ -122,10 +122,8 @@ pub const KV_RATE_LIMIT_HTTP_ANONYMOUS: u8 = 9;
pub const KV_RATE_LIMIT_IMAP: u8 = 10;
pub const KV_QUOTA_BLOB: u8 = 11;
pub const KV_GREYLIST: u8 = 16;
pub const KV_LOCK_PURGE_ACCOUNT: u8 = 20;
pub const KV_LOCK_QUEUE_MESSAGE: u8 = 21;
pub const KV_LOCK_TASK: u8 = 23;
pub const KV_LOCK_HOUSEKEEPER: u8 = 24;
pub const KV_LOCK_DAV: u8 = 25;
pub const KV_SIEVE_ID: u8 = 26;
@@ -272,7 +270,6 @@ pub struct HttpAuthCache {
pub struct Ipc {
pub push_tx: mpsc::Sender<PushEvent>,
pub housekeeper_tx: mpsc::Sender<HousekeeperEvent>,
pub task_tx: Arc<Notify>,
pub queue_tx: mpsc::Sender<QueueEvent>,
pub report_tx: mpsc::Sender<ReportingEvent>,

5
crates/common/src/manager/application.rs
View File

@@ -132,10 +132,11 @@ impl WebApplicationManager {
// Update routes
self.routes.store(routes.into());
trc::event!(
let todo = "use new event";
/*trc::event!(
Resource(trc::ResourceEvent::WebadminUnpacked),
Path = self.bundle_path.path.to_string_lossy().into_owned(),
);
);*/
Ok(())
}

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

@@ -10,10 +10,7 @@ use crate::{
config::{
network::AsnGeoLookupConfig, server::Listeners, storage::Storage, telemetry::Telemetry,
},
ipc::{
BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent,
TrainTaskController,
},
ipc::{BroadcastEvent, PushEvent, QueueEvent, ReportingEvent, TrainTaskController},
};
use arc_swap::ArcSwap;
use pwhash::sha512_crypt;
@@ -39,7 +36,6 @@ pub struct BootManager {
pub struct IpcReceivers {
pub push_rx: Option<mpsc::Receiver<PushEvent>>,
pub housekeeper_rx: Option<mpsc::Receiver<HousekeeperEvent>>,
pub queue_rx: Option<mpsc::Receiver<QueueEvent>>,
pub report_rx: Option<mpsc::Receiver<ReportingEvent>>,
pub broadcast_rx: Option<mpsc::Receiver<BroadcastEvent>>,
@@ -358,14 +354,12 @@ impl BootManager {
pub fn build_ipc(has_pubsub: bool) -> (Ipc, IpcReceivers) {
// Build ipc receivers
let (push_tx, push_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);
let (report_tx, report_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
let (broadcast_tx, broadcast_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
(
Ipc {
push_tx,
housekeeper_tx,
queue_tx,
report_tx,
broadcast_tx: has_pubsub.then_some(broadcast_tx),
@@ -374,7 +368,6 @@ pub fn build_ipc(has_pubsub: bool) -> (Ipc, IpcReceivers) {
},
IpcReceivers {
push_rx: Some(push_rx),
housekeeper_rx: Some(housekeeper_rx),
queue_rx: Some(queue_rx),
report_rx: Some(report_rx),
broadcast_rx: has_pubsub.then_some(broadcast_rx),

8
crates/common/src/telemetry/metrics/store.rs
View File

@@ -61,10 +61,10 @@ impl MetricsStore for Store {
MetricType::ManageSieveConnectionStart,
MetricType::HttpConnectionStart,
MetricType::DeliveryAttemptStart,
MetricType::QueueQueueMessage,
MetricType::QueueQueueMessageAuthenticated,
MetricType::QueueQueueDsn,
MetricType::QueueQueueReport,
MetricType::QueueMessageQueued,
MetricType::QueueAuthenticatedMessageQueued,
MetricType::QueueDsnQueued,
MetricType::QueueReportQueued,
MetricType::MessageIngestHam,
MetricType::MessageIngestSpam,
MetricType::AuthFailed,

10
crates/common/src/telemetry/tracers/store.rs
View File

@@ -222,11 +222,11 @@ impl StoreTracer {
| EventType::Dkim(_)
| EventType::MailAuth(_)
| EventType::Queue(
QueueEvent::QueueMessage
| QueueEvent::QueueMessageAuthenticated
| QueueEvent::QueueReport
| QueueEvent::QueueDsn
| QueueEvent::QueueAutogenerated
QueueEvent::MessageQueued
| QueueEvent::AuthenticatedMessageQueued
| QueueEvent::ReportQueued
| QueueEvent::DsnQueued
| QueueEvent::AutogeneratedQueued
| QueueEvent::Rescheduled
| QueueEvent::RateLimitExceeded
| QueueEvent::ConcurrencyLimitExceeded