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

Nats + Redis PubSub cluster updates replacing Gossip protocol

Browse Source
This commit is contained in:
mdecimus
2025-05-14 19:43:47 +02:00
parent 7ec5701af8
commit 839b7189fa
65 changed files with 1807 additions and 1350 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,24 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{
net::{IpAddr, Ipv4Addr, Ipv6Addr},
sync::Arc,
};
use ahash::{AHashMap, AHashSet};
use arc_swap::ArcSwap;
use mail_auth::{MX, Parameters, Txt};
use mail_send::smtp::tls::build_tls_connector;
use nlp::bayes::{TokenHash, Weights};
use parking_lot::RwLock;
use store::write::BatchBuilder;
use utils::{
cache::{Cache, CacheWithTtl},
config::Config,
snowflake::SnowflakeIdGenerator,
};
use super::server::tls::{build_self_signed_cert, parse_certificates};
use crate::{
CacheSwap, Caches, Data, DavResource, DavResources, MailboxCache, MessageStoreCache,
MessageUidCache, TlsConnectors,
@@ -30,8 +13,21 @@ use crate::{
listener::blocked::BlockedIps,
manager::webadmin::WebAdminManager,
};
use super::server::tls::{build_self_signed_cert, parse_certificates};
use ahash::{AHashMap, AHashSet};
use arc_swap::ArcSwap;
use mail_auth::{MX, Parameters, Txt};
use mail_send::smtp::tls::build_tls_connector;
use nlp::bayes::{TokenHash, Weights};
use parking_lot::RwLock;
use std::{
net::{IpAddr, Ipv4Addr, Ipv6Addr},
sync::Arc,
};
use utils::{
cache::{Cache, CacheWithTtl},
config::Config,
snowflake::{HlcTimestamp, SnowflakeIdGenerator},
};
impl Data {
pub fn parse(config: &mut Config) -> Self {
@@ -48,7 +44,7 @@ impl Data {
.property::<u64>("cluster.node-id")
.unwrap_or_else(store::rand::random);
let id_generator = SnowflakeIdGenerator::with_node_id(node_id);
BatchBuilder::init_id_generator(node_id as u16);
HlcTimestamp::init(node_id as u16);
if !id_generator.is_valid() {
panic!("Invalid system time, panicking to avoid data corruption");
}
@@ -65,7 +61,6 @@ impl Data {
.ok()
.map(Arc::new),
blocked_ips: RwLock::new(BlockedIps::parse(config).blocked_ip_addresses),
blocked_ips_version: 0.into(),
jmap_id_gen: id_generator.clone(),
queue_id_gen: id_generator.clone(),
span_id_gen: id_generator,
@@ -74,7 +69,6 @@ impl Data {
.value("webadmin.path")
.map(|path| WebAdminManager::new(path.into()))
.unwrap_or_default(),
config_version: 0.into(),
logos: Default::default(),
smtp_connectors: TlsConnectors::default(),
asn_geo_data: Default::default(),
@@ -225,13 +219,11 @@ impl Default for Data {
tls_certificates: Default::default(),
tls_self_signed_cert: Default::default(),
blocked_ips: Default::default(),
blocked_ips_version: 0.into(),
jmap_id_gen: Default::default(),
queue_id_gen: Default::default(),
span_id_gen: Default::default(),
queue_status: true.into(),
webadmin: Default::default(),
config_version: Default::default(),
logos: Default::default(),
smtp_connectors: Default::default(),
asn_geo_data: Default::default(),

16
crates/common/src/config/mod.rs
View File

@@ -137,6 +137,21 @@ impl Core {
}
})
.unwrap_or_default();
let pubsub = config
.value("storage.pubsub")
.map(|id| id.to_string())
.and_then(|id| {
if let Some(store) = stores.pubsub_stores.get(&id) {
store.clone().into()
} else {
config.new_parse_error(
"storage.pubsub",
format!("PubSub store {id:?} not found"),
);
None
}
})
.unwrap_or_default();
let mut directories =
Directories::parse(config, &stores, data.clone(), is_enterprise).await;
let directory = config
@@ -192,6 +207,7 @@ impl Core {
blob,
fts,
lookup,
pubsub,
directory,
directories: directories.directories,
purge_schedules: stores.purge_schedules,

3
crates/common/src/config/storage.rs
View File

@@ -8,7 +8,7 @@ use std::sync::Arc;
use ahash::AHashMap;
use directory::Directory;
use store::{BlobStore, FtsStore, InMemoryStore, PurgeSchedule, Store};
use store::{BlobStore, FtsStore, InMemoryStore, PubSubStore, PurgeSchedule, Store};
use crate::manager::config::ConfigManager;
@@ -18,6 +18,7 @@ pub struct Storage {
pub blob: BlobStore,
pub fts: FtsStore,
pub lookup: InMemoryStore,
pub pubsub: PubSubStore,
pub directory: Arc<Directory>,
pub directories: AHashMap<String, Arc<Directory>>,
pub purge_schedules: Vec<PurgeSchedule>,

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

@@ -4,8 +4,15 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{sync::Arc, time::Duration};
use crate::{
Inner, Server,
auth::{AccessToken, ResourceToken, TenantInfo},
config::smtp::{
auth::{ArcSealer, DkimSigner, LazySignature, ResolvedSignature, build_signature},
queue::RelayHost,
},
ipc::{BroadcastEvent, StateEvent},
};
use directory::{Directory, QueryBy, Type, backend::internal::manage::ManageDirectory};
use jmap_proto::types::{
blob::BlobId,
@@ -15,6 +22,7 @@ use jmap_proto::types::{
type_state::DataType,
};
use sieve::Sieve;
use std::{sync::Arc, time::Duration};
use store::{
BitmapKey, BlobClass, BlobStore, Deserialize, FtsStore, InMemoryStore, IndexKey, IterateParams,
LogKey, SerializeInfallible, Store, U32_LEN, ValueKey,
@@ -28,16 +36,6 @@ use store::{
use trc::AddContext;
use utils::BlobHash;
use crate::{
Inner, Server,
auth::{AccessToken, ResourceToken, TenantInfo},
config::smtp::{
auth::{ArcSealer, DkimSigner, LazySignature, ResolvedSignature, build_signature},
queue::RelayHost,
},
ipc::StateEvent,
};
impl Server {
#[inline(always)]
pub fn store(&self) -> &Store {
@@ -510,22 +508,21 @@ impl Server {
if let Some(changes) = builder.changes() {
for (account_id, changed_collections) in changes {
let mut state_change = StateChange::new(account_id);
let change_id = changed_collections.change_id;
let mut state_change = StateChange::new(account_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);
state_change.set_change(data_type);
}
}
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);
state_change.set_change(data_type);
}
}
if state_change.has_changes() {
self.broadcast_state_change(state_change).await;
}
assigned_ids.change_id = change_id.into();
assigned_ids.change_id = changed_collections.change_id.into();
}
}
@@ -576,7 +573,10 @@ impl Server {
.ipc
.state_tx
.clone()
.send(StateEvent::Publish { state_change })
.send(StateEvent::Publish {
state_change,
broadcast: true,
})
.await
{
Ok(_) => true,
@@ -592,6 +592,18 @@ impl Server {
}
}
pub async fn cluster_broadcast(&self, event: BroadcastEvent) {
if let Some(broadcast_tx) = &self.inner.ipc.broadcast_tx.clone() {
if broadcast_tx.send(event).await.is_err() {
trc::event!(
Server(trc::ServerEvent::ThreadError),
Details = "Error sending broadcast event.",
CausedBy = trc::location!()
);
}
}
}
#[allow(clippy::blocks_in_conditions)]
pub async fn put_blob(
&self,
@@ -655,13 +667,6 @@ impl Server {
})
}
pub fn increment_config_version(&self) {
self.inner
.data
.config_version
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
pub async fn total_accounts(&self) -> trc::Result<u64> {
self.store()
.count_principals(None, Type::Individual.into(), None)

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

@@ -54,6 +54,7 @@ pub enum StateEvent {
},
Publish {
state_change: StateChange,
broadcast: bool,
},
UpdateSharedAccounts {
account_id: u32,
@@ -65,6 +66,13 @@ pub enum StateEvent {
Stop,
}
#[derive(Debug)]
pub enum BroadcastEvent {
StateChange(StateChange),
ReloadSettings,
ReloadBlockedIps,
}
#[derive(Debug)]
pub enum UpdateSubscription {
Unverified {

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

@@ -24,7 +24,7 @@ use config::{
storage::Storage,
telemetry::Metrics,
};
use ipc::{HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent};
use ipc::{BroadcastEvent, HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent};
use jmap_proto::types::value::AclGrant;
use listener::{asn::AsnGeoLookupData, blocked::Security, tls::AcmeProviders};
use mail_auth::{MX, Txt};
@@ -35,10 +35,7 @@ use rustls::sign::CertifiedKey;
use std::{
hash::{BuildHasher, Hash, Hasher},
net::{IpAddr, Ipv4Addr, Ipv6Addr},
sync::{
Arc,
atomic::{AtomicBool, AtomicU8},
},
sync::{Arc, atomic::AtomicBool},
time::Duration,
};
use store::roaring::RoaringBitmap;
@@ -126,7 +123,6 @@ pub struct Data {
pub tls_self_signed_cert: Option<Arc<CertifiedKey>>,
pub blocked_ips: RwLock<AHashSet<IpAddr>>,
pub blocked_ips_version: AtomicU8,
pub asn_geo_data: AsnGeoLookupData,
@@ -137,7 +133,6 @@ pub struct Data {
pub webadmin: WebAdminManager,
pub logos: Mutex<AHashMap<String, Option<Resource<Vec<u8>>>>>,
pub config_version: AtomicU8,
pub smtp_connectors: TlsConnectors,
}
@@ -233,6 +228,7 @@ pub struct Ipc {
pub index_tx: Arc<Notify>,
pub queue_tx: mpsc::Sender<QueueEvent>,
pub report_tx: mpsc::Sender<ReportingEvent>,
pub broadcast_tx: Option<mpsc::Sender<BroadcastEvent>>,
pub local_delivery_sm: Arc<Semaphore>,
}
@@ -476,6 +472,7 @@ impl Default for Ipc {
index_tx: Default::default(),
queue_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
report_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0,
broadcast_tx: None,
local_delivery_sm: Arc::new(Semaphore::new(10)),
}
}

14
crates/common/src/listener/blocked.rs
View File

@@ -18,7 +18,7 @@ use utils::{
use crate::{
KV_RATE_LIMIT_AUTH, KV_RATE_LIMIT_LOITER, KV_RATE_LIMIT_RCPT, KV_RATE_LIMIT_SCAN, Server,
ip_to_bytes, manager::config::MatchType,
ip_to_bytes, ipc::BroadcastEvent, manager::config::MatchType,
};
#[derive(Debug, Clone)]
@@ -226,7 +226,7 @@ impl Server {
Ok(false)
}
async fn block_ip(&self, ip: IpAddr) -> trc::Result<()> {
pub async fn block_ip(&self, ip: IpAddr) -> trc::Result<()> {
// Add IP to blocked list
self.inner.data.blocked_ips.write().insert(ip);
@@ -244,7 +244,8 @@ impl Server {
.await?;
// Increment version
self.increment_blocked_version();
self.cluster_broadcast(BroadcastEvent::ReloadBlockedIps)
.await;
Ok(())
}
@@ -276,13 +277,6 @@ impl Server {
.iter()
.any(|network| network.matches(ip)))
}
pub fn increment_blocked_version(&self) {
self.inner
.data
.blocked_ips_version
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
}
impl BlockedIps {

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

@@ -27,7 +27,7 @@ use crate::{
Caches, Core, Data, IPC_CHANNEL_BUFFER, Inner, Ipc,
config::{network::AsnGeoLookupConfig, server::Listeners, telemetry::Telemetry},
core::BuildServer,
ipc::{HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent},
ipc::{BroadcastEvent, HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent},
};
use super::{
@@ -49,6 +49,7 @@ pub struct IpcReceivers {
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>>,
}
const HELP: &str = concat!(
@@ -424,7 +425,7 @@ impl BootManager {
core.network.asn_geo_lookup,
AsnGeoLookupConfig::Resource { .. }
);
let (ipc, ipc_rxs) = build_ipc(&mut config);
let (ipc, ipc_rxs) = build_ipc(&mut config, !core.storage.pubsub.is_none());
let inner = Arc::new(Inner {
shared_core: ArcSwap::from_pointee(core),
data,
@@ -487,18 +488,20 @@ impl BootManager {
}
}
pub fn build_ipc(config: &mut Config) -> (Ipc, IpcReceivers) {
pub fn build_ipc(config: &mut Config, has_pubsub: bool) -> (Ipc, IpcReceivers) {
// Build ipc receivers
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);
let (report_tx, report_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
let (broadcast_tx, broadcast_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
(
Ipc {
state_tx,
housekeeper_tx,
queue_tx,
report_tx,
broadcast_tx: has_pubsub.then_some(broadcast_tx),
index_tx: Arc::new(Notify::new()),
local_delivery_sm: Arc::new(Semaphore::new(
config
@@ -512,6 +515,7 @@ pub fn build_ipc(config: &mut Config) -> (Ipc, IpcReceivers) {
housekeeper_rx: Some(housekeeper_rx),
queue_rx: Some(queue_rx),
report_rx: Some(report_rx),
broadcast_rx: has_pubsub.then_some(broadcast_rx),
},
)
}

1
crates/common/src/manager/reload.rs
View File

@@ -76,6 +76,7 @@ impl Server {
blob_stores: self.core.storage.blobs.clone(),
fts_stores: self.core.storage.ftss.clone(),
in_memory_stores: self.core.storage.lookups.clone(),
pubsub_stores: Default::default(),
purge_schedules: Default::default(),
};
stores.parse_stores(&mut config).await;