diff --git a/Cargo.lock b/Cargo.lock index 5c9dd32f..b05aed7d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1263,7 +1263,7 @@ dependencies = [ [[package]] name = "common" -version = "0.12.5" +version = "0.13.0" dependencies = [ "aes-gcm-siv", "ahash", @@ -1733,7 +1733,7 @@ checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476" [[package]] name = "dav" -version = "0.12.5" +version = "0.13.0" dependencies = [ "calcard", "chrono", @@ -1755,7 +1755,7 @@ dependencies = [ [[package]] name = "dav-proto" -version = "0.12.5" +version = "0.13.0" dependencies = [ "calcard", "chrono", @@ -1948,7 +1948,7 @@ dependencies = [ [[package]] name = "directory" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "argon2", @@ -2226,7 +2226,7 @@ dependencies = [ [[package]] name = "email" -version = "0.12.5" +version = "0.13.0" dependencies = [ "aes", "aes-gcm", @@ -2356,7 +2356,7 @@ dependencies = [ [[package]] name = "event_macro" -version = "0.12.5" +version = "0.13.0" dependencies = [ "proc-macro2", "quote", @@ -2842,7 +2842,7 @@ dependencies = [ [[package]] name = "groupware" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "calcard", @@ -3142,7 +3142,7 @@ dependencies = [ [[package]] name = "http" -version = "0.12.5" +version = "0.13.0" dependencies = [ "async-stream", "base64 0.22.1", @@ -3252,7 +3252,7 @@ dependencies = [ [[package]] name = "http_proto" -version = "0.12.5" +version = "0.13.0" dependencies = [ "common", "compact_str", @@ -3554,7 +3554,7 @@ checksum = "09e54e57b4c48b40f7aec75635392b12b3421fa26fe8b4332e63138ed278459c" [[package]] name = "imap" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "common", @@ -3581,7 +3581,7 @@ dependencies = [ [[package]] name = "imap_proto" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "chrono", @@ -3827,7 +3827,7 @@ dependencies = [ [[package]] name = "jmap" -version = "0.12.5" +version = "0.13.0" dependencies = [ "aes-gcm", "aes-gcm-siv", @@ -3898,7 +3898,7 @@ dependencies = [ [[package]] name = "jmap_proto" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "compact_str", @@ -4372,7 +4372,7 @@ dependencies = [ [[package]] name = "managesieve" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "common", @@ -4474,7 +4474,7 @@ checksum = "c797b9d6bb23aab2fc369c65f871be49214f5c759af65bde26ffaaa2b646b492" [[package]] name = "migration" -version = "0.12.5" +version = "0.13.0" dependencies = [ "base64 0.22.1", "bincode 1.3.3", @@ -4712,7 +4712,7 @@ dependencies = [ [[package]] name = "nlp" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "bincode 1.3.3", @@ -5440,7 +5440,7 @@ dependencies = [ [[package]] name = "pop3" -version = "0.12.5" +version = "0.13.0" dependencies = [ "common", "directory", @@ -5607,7 +5607,7 @@ dependencies = [ [[package]] name = "proc_macros" -version = "0.12.5" +version = "0.13.0" dependencies = [ "proc-macro2", "quote", @@ -7271,7 +7271,7 @@ dependencies = [ [[package]] name = "services" -version = "0.12.5" +version = "0.13.0" dependencies = [ "aes-gcm", "aes-gcm-siv", @@ -7483,7 +7483,7 @@ checksum = "8917285742e9f3e1683f0a9c4e6b57960b7314d0b08d30d1ecd426713ee2eee9" [[package]] name = "smtp" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "blake3", @@ -7574,7 +7574,7 @@ dependencies = [ [[package]] name = "spam-filter" -version = "0.12.5" +version = "0.13.0" dependencies = [ "common", "compact_str", @@ -7638,14 +7638,14 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" [[package]] name = "stalwart" -version = "0.12.5" +version = "0.13.0" dependencies = [ "common", "dav", "directory", "email", "groupware", - "http 0.12.5", + "http 0.13.0", "imap", "jemallocator", "jmap", @@ -7664,7 +7664,7 @@ dependencies = [ [[package]] name = "stalwart-cli" -version = "0.12.5" +version = "0.13.0" dependencies = [ "clap", "console", @@ -7695,7 +7695,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "store" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "arc-swap", @@ -7904,7 +7904,7 @@ dependencies = [ [[package]] name = "tests" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "async-trait", @@ -7925,7 +7925,7 @@ dependencies = [ "form_urlencoded", "futures", "groupware", - "http 0.12.5", + "http 0.13.0", "http-body-util", "http_proto", "hyper 1.6.0", @@ -8451,7 +8451,7 @@ dependencies = [ [[package]] name = "trc" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "base64 0.22.1", @@ -8757,7 +8757,7 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "utils" -version = "0.12.5" +version = "0.13.0" dependencies = [ "ahash", "base64 0.22.1", diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index 10c9c5d1..5233f541 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -5,7 +5,7 @@ authors = ["Stalwart Labs LLC "] license = "AGPL-3.0-only OR LicenseRef-SEL" repository = "https://github.com/stalwartlabs/cli" homepage = "https://github.com/stalwartlabs/cli" -version = "0.12.5" +version = "0.13.0" edition = "2024" readme = "README.md" resolver = "2" diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml index 045dcb88..89999c70 100644 --- a/crates/common/Cargo.toml +++ b/crates/common/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "common" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" build = "build.rs" diff --git a/crates/common/src/config/smtp/mod.rs b/crates/common/src/config/smtp/mod.rs index 199ea1df..9dfafa36 100644 --- a/crates/common/src/config/smtp/mod.rs +++ b/crates/common/src/config/smtp/mod.rs @@ -98,7 +98,7 @@ pub(crate) const SMTP_RCPT_TO_VARS: &[u32; 17] = &[ V_ASN, V_COUNTRY, ]; -pub(crate) const SMTP_QUEUE_HOST_VARS: &[u32; 14] = &[ +pub(crate) const SMTP_QUEUE_HOST_VARS: &[u32; 20] = &[ V_SENDER, V_SENDER_DOMAIN, V_RECIPIENT_DOMAIN, @@ -113,8 +113,14 @@ pub(crate) const SMTP_QUEUE_HOST_VARS: &[u32; 14] = &[ V_QUEUE_EXPIRES_IN, V_QUEUE_LAST_STATUS, V_QUEUE_LAST_ERROR, + V_QUEUE_NAME, + V_QUEUE_AGE, + V_RECEIVED_FROM_IP, + V_RECEIVED_VIA_PORT, + V_SOURCE, + V_SIZE, ]; -pub(crate) const SMTP_QUEUE_RCPT_VARS: &[u32; 13] = &[ +pub(crate) const SMTP_QUEUE_RCPT_VARS: &[u32; 17] = &[ V_RECIPIENT, V_RECIPIENT_DOMAIN, V_RECIPIENTS, @@ -126,8 +132,12 @@ pub(crate) const SMTP_QUEUE_RCPT_VARS: &[u32; 13] = &[ V_QUEUE_EXPIRES_IN, V_QUEUE_LAST_STATUS, V_QUEUE_LAST_ERROR, - V_REMOTE_IP, - V_LOCAL_PORT, + V_QUEUE_NAME, + V_QUEUE_AGE, + V_RECEIVED_FROM_IP, + V_RECEIVED_VIA_PORT, + V_SOURCE, + V_SIZE, ]; pub(crate) const SMTP_QUEUE_SENDER_VARS: &[u32; 8] = &[ V_SENDER, diff --git a/crates/common/src/config/smtp/queue.rs b/crates/common/src/config/smtp/queue.rs index aeefee42..eacd57bd 100644 --- a/crates/common/src/config/smtp/queue.rs +++ b/crates/common/src/config/smtp/queue.rs @@ -86,7 +86,7 @@ pub struct Dsn { #[derive(Clone, Debug)] pub struct VirtualQueue { - pub threads: u32, + pub threads: usize, } #[derive(Clone, Debug)] @@ -252,7 +252,6 @@ impl QueueConfig { *value = if_block; } } - let todo = "test parsing"; // Parse strategies queue.virtual_queues = parse_virtual_queues(config); @@ -369,7 +368,7 @@ fn parse_virtual_queues(config: &mut Config) -> AHashMap Option { Some(VirtualQueue { threads: config - .property_require::(("queue.virtual", id, "threads-per-node")) + .property_require::(("queue.virtual", id, "threads-per-node")) .unwrap_or(1), }) } @@ -919,7 +918,9 @@ impl QueueName { } pub fn as_str(&self) -> &str { - std::str::from_utf8(&self.0).unwrap_or_default() + std::str::from_utf8(&self.0) + .unwrap_or_default() + .trim_end_matches('\0') } pub fn into_inner(self) -> [u8; 8] { diff --git a/crates/common/src/core.rs b/crates/common/src/core.rs index 209af3e2..93974e89 100644 --- a/crates/common/src/core.rs +++ b/crates/common/src/core.rs @@ -217,7 +217,7 @@ impl Server { }) } - pub fn get_virtual_queue_or_default(&self, name: &QueueName, session_id: u64) -> &VirtualQueue { + pub fn get_virtual_queue_or_default(&self, name: &QueueName) -> &VirtualQueue { static DEFAULT_QUEUE: VirtualQueue = VirtualQueue { threads: 25 }; self.core .smtp @@ -230,7 +230,6 @@ impl Server { Smtp(trc::SmtpEvent::IdNotFound), Id = name.to_string(), Details = "Virtual queue not found", - SpanId = session_id, ); } diff --git a/crates/common/src/expr/mod.rs b/crates/common/src/expr/mod.rs index 740368a5..5e07c278 100644 --- a/crates/common/src/expr/mod.rs +++ b/crates/common/src/expr/mod.rs @@ -38,6 +38,12 @@ pub const V_HEADERS: u32 = 23; pub const V_METHOD: u32 = 24; pub const V_ASN: u32 = 25; pub const V_COUNTRY: u32 = 26; +pub const V_RECEIVED_VIA_PORT: u32 = 27; +pub const V_RECEIVED_FROM_IP: u32 = 28; +pub const V_QUEUE_NAME: u32 = 29; +pub const V_SOURCE: u32 = 30; +pub const V_SIZE: u32 = 31; +pub const V_QUEUE_AGE: u32 = 32; pub const VARIABLES_MAP: &[(&str, u32)] = &[ ("rcpt", V_RECIPIENT), @@ -67,6 +73,12 @@ pub const VARIABLES_MAP: &[(&str, u32)] = &[ ("method", V_METHOD), ("asn", V_ASN), ("country", V_COUNTRY), + ("received_via_port", V_RECEIVED_VIA_PORT), + ("received_from_ip", V_RECEIVED_FROM_IP), + ("queue_name", V_QUEUE_NAME), + ("source", V_SOURCE), + ("size", V_SIZE), + ("queue_age", V_QUEUE_AGE), ]; use compact_str::CompactString; diff --git a/crates/common/src/ipc.rs b/crates/common/src/ipc.rs index 421cff5b..09c3dc9d 100644 --- a/crates/common/src/ipc.rs +++ b/crates/common/src/ipc.rs @@ -18,6 +18,7 @@ use tokio::sync::mpsc; use utils::map::bitmap::Bitmap; use crate::config::smtp::{ + queue::QueueName, report::AggregateFrequency, resolver::{Policy, Tlsa}, }; @@ -104,16 +105,18 @@ pub enum QueueEvent { Refresh, WorkerDone { queue_id: u64, + queue_name: QueueName, status: QueueEventStatus, }, Paused(bool), + ReloadSettings, Stop, } #[derive(Debug)] pub enum QueueEventStatus { Completed, - Locked { until: u64 }, + Locked, Deferred, } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index a17d9f64..2f3af0ca 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -72,7 +72,17 @@ pub static USER_AGENT: &str = "Stalwart/1.0.0"; pub static DAEMON_NAME: &str = concat!("Stalwart v", env!("CARGO_PKG_VERSION"),); pub static PROD_ID: &str = "-//Stalwart Labs LLC//Stalwart Server//EN"; -pub const DATABASE_SCHEMA_VERSION: u32 = 2; +/* + +Schema history: + +1 - v0.12.0 +2 - v0.12.4 +3 - v0.13.0 + +*/ + +pub const DATABASE_SCHEMA_VERSION: u32 = 3; pub const LONG_1D_SLUMBER: Duration = Duration::from_secs(60 * 60 * 24); pub const LONG_1Y_SLUMBER: Duration = Duration::from_secs(60 * 60 * 24 * 365); diff --git a/crates/dav-proto/Cargo.toml b/crates/dav-proto/Cargo.toml index b156b0f2..ab901a60 100644 --- a/crates/dav-proto/Cargo.toml +++ b/crates/dav-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dav-proto" -version = "0.12.5" +version = "0.13.0" edition = "2021" [dependencies] diff --git a/crates/dav/Cargo.toml b/crates/dav/Cargo.toml index cd147377..6b74b7f8 100644 --- a/crates/dav/Cargo.toml +++ b/crates/dav/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dav" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/directory/Cargo.toml b/crates/directory/Cargo.toml index c12e2d71..70400096 100644 --- a/crates/directory/Cargo.toml +++ b/crates/directory/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "directory" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/email/Cargo.toml b/crates/email/Cargo.toml index 6c01ee31..07582490 100644 --- a/crates/email/Cargo.toml +++ b/crates/email/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "email" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/groupware/Cargo.toml b/crates/groupware/Cargo.toml index 9329b2dc..5d3d2887 100644 --- a/crates/groupware/Cargo.toml +++ b/crates/groupware/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "groupware" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/http-proto/Cargo.toml b/crates/http-proto/Cargo.toml index 1ea2a4d5..76825549 100644 --- a/crates/http-proto/Cargo.toml +++ b/crates/http-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "http_proto" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/http/Cargo.toml b/crates/http/Cargo.toml index 5432fd46..293dbac4 100644 --- a/crates/http/Cargo.toml +++ b/crates/http/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "http" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/imap-proto/Cargo.toml b/crates/imap-proto/Cargo.toml index 99cf7cec..91aaf862 100644 --- a/crates/imap-proto/Cargo.toml +++ b/crates/imap-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "imap_proto" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/imap/Cargo.toml b/crates/imap/Cargo.toml index c5a8eaf7..8bba0b31 100644 --- a/crates/imap/Cargo.toml +++ b/crates/imap/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "imap" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/jmap-proto/Cargo.toml b/crates/jmap-proto/Cargo.toml index 2ba62e9e..3738fd36 100644 --- a/crates/jmap-proto/Cargo.toml +++ b/crates/jmap-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "jmap_proto" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index f365d1c7..4faaf47e 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "jmap" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index f4b05a0a..252e8236 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -7,7 +7,7 @@ homepage = "https://stalw.art" keywords = ["imap", "jmap", "smtp", "email", "mail", "webdav", "server"] categories = ["email"] license = "AGPL-3.0-only OR LicenseRef-SEL" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/managesieve/Cargo.toml b/crates/managesieve/Cargo.toml index 08f622c4..27f2b5a3 100644 --- a/crates/managesieve/Cargo.toml +++ b/crates/managesieve/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "managesieve" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/migration/Cargo.toml b/crates/migration/Cargo.toml index 1c2ecda1..a1ab2bbc 100644 --- a/crates/migration/Cargo.toml +++ b/crates/migration/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "migration" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/migration/src/lib.rs b/crates/migration/src/lib.rs index d8710970..768ee5bc 100644 --- a/crates/migration/src/lib.rs +++ b/crates/migration/src/lib.rs @@ -4,25 +4,25 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::calendar::migrate_calendar_events; +use crate::{ + calendar::migrate_calendar_events, + queue::{migrate_queue_v011, migrate_queue_v012}, + tasks::migrate_tasks_v011, +}; use changelog::reset_changelog; use common::{DATABASE_SCHEMA_VERSION, KV_LOCK_HOUSEKEEPER, Server}; use jmap_proto::types::{collection::Collection, property::Property}; use principal::{migrate_principal, migrate_principals}; -use queue::migrate_queue; use report::migrate_reports; use std::time::Duration; use store::{ Deserialize, IterateParams, SUBSPACE_PROPERTY, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_REPORT_IN, - SUBSPACE_REPORT_OUT, SUBSPACE_TASK_QUEUE, SerializeInfallible, U32_LEN, Value, ValueKey, + SUBSPACE_REPORT_OUT, SerializeInfallible, U32_LEN, Value, ValueKey, dispatch::{DocumentSet, lookup::KeyValue}, rand::{self, seq::SliceRandom}, - write::{ - AnyClass, AnyKey, BatchBuilder, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now, - }, + write::{AnyClass, AnyKey, BatchBuilder, ValueClass, key::DeserializeBigEndian}, }; use trc::AddContext; -use utils::BlobHash; pub mod calendar; pub mod changelog; @@ -37,6 +37,7 @@ pub mod queue; pub mod report; pub mod sieve; pub mod submission; +pub mod tasks; pub mod threads; const LOCK_WAIT_TIME_ACCOUNT: u64 = 3 * 60; @@ -45,7 +46,14 @@ const LOCK_RETRY_TIME: Duration = Duration::from_secs(30); pub async fn try_migrate(server: &Server) -> trc::Result<()> { if std::env::var("FORCE_MIGRATE_QUEUE").is_ok() { - migrate_queue(server).await.caused_by(trc::location!())?; + migrate_queue_v012(server) + .await + .caused_by(trc::location!())?; + return Ok(()); + } else if std::env::var("FORCE_MIGRATE_QUEUE_V011").is_ok() { + migrate_queue_v011(server) + .await + .caused_by(trc::location!())?; return Ok(()); } else if let Some(account_id) = std::env::var("FORCE_MIGRATE_ACCOUNT") .ok() @@ -57,9 +65,6 @@ pub async fn try_migrate(server: &Server) -> trc::Result<()> { return Ok(()); } - let todo = - "migrate queue + new LegacyRecipient with domain_idx u32 / size u64 + migrate error enum"; - match server .store() .get_value::(AnyKey { @@ -73,7 +78,14 @@ pub async fn try_migrate(server: &Server) -> trc::Result<()> { return Ok(()); } Some(1) => { - migrate_v0_12_0(server).await.caused_by(trc::location!())?; + migrate_v0_12(server, true) + .await + .caused_by(trc::location!())?; + } + Some(2) => { + migrate_v0_12(server, false) + .await + .caused_by(trc::location!())?; } Some(version) => { panic!( @@ -105,7 +117,7 @@ pub async fn try_migrate(server: &Server) -> trc::Result<()> { Ok(()) } -async fn migrate_v0_12_0(server: &Server) -> trc::Result<()> { +async fn migrate_v0_12(server: &Server, migrate_tasks: bool) -> trc::Result<()> { let force_lock = std::env::var("FORCE_LOCK").is_ok(); let in_memory = server.in_memory_store(); @@ -120,76 +132,14 @@ async fn migrate_v0_12_0(server: &Server) -> trc::Result<()> { .await .caused_by(trc::location!())? { - let from_key = ValueKey:: { - account_id: 0, - collection: 0, - document_id: 0, - class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { - due: 0, - hash: BlobHash::default(), - }), - }; - let to_key = ValueKey:: { - account_id: u32::MAX, - collection: u8::MAX, - document_id: u32::MAX, - class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { - due: u64::MAX, - hash: BlobHash::default(), - }), - }; - - let now = now(); - let mut migrate_tasks = Vec::new(); - server - .core - .storage - .data - .iterate( - IterateParams::new(from_key, to_key).ascending(), - |key, value| { - let due = key.deserialize_be_u64(0)?; - - if due > now { - migrate_tasks.push((key.to_vec(), value.to_vec())); - } - - Ok(true) - }, - ) + migrate_queue_v012(server) .await .caused_by(trc::location!())?; - if !migrate_tasks.is_empty() { - let num_migrated = migrate_tasks.len(); - let mut batch = BatchBuilder::new(); - for (key, value) in migrate_tasks { - let mut new_key = key.clone(); - new_key[0..8].copy_from_slice(&now.to_be_bytes()); - - batch - .clear(ValueClass::Any(AnyClass { - subspace: SUBSPACE_TASK_QUEUE, - key, - })) - .set( - ValueClass::Any(AnyClass { - subspace: SUBSPACE_TASK_QUEUE, - key: new_key, - }), - value, - ); - } - server - .store() - .write(batch.build_all()) + if migrate_tasks { + migrate_tasks_v011(server) .await .caused_by(trc::location!())?; - - trc::event!( - Server(trc::ServerEvent::Startup), - Details = format!("Migrated {num_migrated} tasks") - ); } in_memory @@ -237,7 +187,9 @@ async fn migrate_v0_11(server: &Server) -> trc::Result<()> { .caused_by(trc::location!())? .is_none() { - migrate_queue(server).await.caused_by(trc::location!())?; + migrate_queue_v011(server) + .await + .caused_by(trc::location!())?; migrate_reports(server).await.caused_by(trc::location!())?; reset_changelog(server).await.caused_by(trc::location!())?; principal_ids = migrate_principals(server) diff --git a/crates/migration/src/queue.rs b/crates/migration/src/queue.rs index 881c5197..b8dbd1f6 100644 --- a/crates/migration/src/queue.rs +++ b/crates/migration/src/queue.rs @@ -25,7 +25,7 @@ use store::{ use trc::AddContext; use utils::BlobHash; -pub(crate) async fn migrate_queue(server: &Server) -> trc::Result<()> { +pub(crate) async fn migrate_queue_v011(server: &Server) -> trc::Result<()> { let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent( store::write::QueueEvent { due: 0, @@ -75,7 +75,7 @@ pub(crate) async fn migrate_queue(server: &Server) -> trc::Result<()> { for queue_id in queue_ids { match server .store() - .get_value::>(ValueKey::from(ValueClass::Queue( + .get_value::>(ValueKey::from(ValueClass::Queue( QueueClass::Message(queue_id), ))) .await @@ -123,8 +123,123 @@ pub(crate) async fn migrate_queue(server: &Server) -> trc::Result<()> { Ok(()) } -impl From for Message { - fn from(message: LegacyMessage) -> Self { +pub(crate) async fn migrate_queue_v012(server: &Server) -> trc::Result<()> { + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent( + store::write::QueueEvent { + due: 0, + queue_id: 0, + queue_name: [0; 8], + }, + ))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent( + store::write::QueueEvent { + due: u64::MAX, + queue_id: u64::MAX, + queue_name: [u8::MAX; 8], + }, + ))); + + let mut queue_ids = AHashSet::new(); + server + .store() + .iterate( + IterateParams::new(from_key, to_key).ascending().no_values(), + |key, _| { + queue_ids.insert(key.deserialize_be_u64(U64_LEN)?); + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(0))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX))); + server + .store() + .iterate( + IterateParams::new(from_key, to_key).ascending().no_values(), + |key, _| { + queue_ids.insert(key.deserialize_be_u64(0)?); + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + let mut count = 0; + + for queue_id in queue_ids { + match server + .store() + .get_value::>(ValueKey::from(ValueClass::Queue( + QueueClass::Message(queue_id), + ))) + .await + .and_then(|archive| { + if let Some(archive) = archive { + archive.deserialize::().map(Some) + } else { + Ok(None) + } + }) { + Ok(Some(archive)) => { + let mut batch = BatchBuilder::new(); + batch.set( + ValueClass::Queue(QueueClass::Message(queue_id)), + Archiver::new(Message::from(archive)) + .serialize() + .caused_by(trc::location!())?, + ); + count += 1; + server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + } + Ok(None) => (), + Err(err) => { + if server + .store() + .get_value::>(ValueKey::from(ValueClass::Queue( + QueueClass::Message(queue_id), + ))) + .await + .and_then(|archive| { + if let Some(archive) = archive { + archive.deserialize::().map(Some) + } else { + Ok(None) + } + }) + .is_err() + { + return Err(err + .ctx(trc::Key::QueueId, queue_id) + .caused_by(trc::location!())); + } + } + } + } + + if count > 0 { + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migrated {count} queued messages",) + ); + } + + Ok(()) +} + +impl From> for Message +where + SIZE: AsU64, + IDX: AsU64, +{ + fn from(message: LegacyMessage) -> Self { let domains = message.domains; Message { created: message.created, @@ -136,7 +251,7 @@ impl From for Message { .recipients .into_iter() .map(|r| { - let domain = &domains[r.domain_idx]; + let domain = &domains[r.domain_idx.as_u64() as usize]; Recipient { address: r.address, address_lcase: r.address_lcase, @@ -170,7 +285,7 @@ impl From for Message { flags: message.flags, env_id: message.env_id, priority: message.priority, - size: message.size as u64, + size: message.size.as_u64(), quota_keys: message.quota_keys, received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), received_via_port: 0, @@ -178,6 +293,25 @@ impl From for Message { } } +trait AsU64 { + fn as_u64(&self) -> u64; +} +impl AsU64 for usize { + fn as_u64(&self) -> u64 { + *self as u64 + } +} +impl AsU64 for u32 { + fn as_u64(&self) -> u64 { + *self as u64 + } +} +impl AsU64 for u64 { + fn as_u64(&self) -> u64 { + *self + } +} + fn migrate_legacy_error(domain: &str, err: &LegacyError) -> ErrorDetails { match err { LegacyError::DnsError(err) => ErrorDetails { @@ -232,8 +366,20 @@ fn migrate_host_response(response: HostResponse) -> ErrorDet } } -#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)] -pub struct LegacyMessage { +pub type MessageV011 = LegacyMessage; +pub type MessageV012 = LegacyMessage; + +#[derive( + Debug, + Clone, + PartialEq, + Eq, + rkyv::Serialize, + rkyv::Deserialize, + rkyv::Archive, + serde::Deserialize, +)] +pub struct LegacyMessage { pub queue_id: QueueId, pub created: u64, pub blob_hash: BlobHash, @@ -241,23 +387,32 @@ pub struct LegacyMessage { pub return_path: String, pub return_path_lcase: String, pub return_path_domain: String, - pub recipients: Vec, + pub recipients: Vec>, pub domains: Vec, pub flags: u64, pub env_id: Option, pub priority: i16, - pub size: usize, + pub size: SIZE, pub quota_keys: Vec, #[serde(skip)] pub span_id: u64, } -#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)] -pub struct LegacyRecipient { - pub domain_idx: usize, +#[derive( + Debug, + Clone, + PartialEq, + Eq, + rkyv::Serialize, + rkyv::Deserialize, + rkyv::Archive, + serde::Deserialize, +)] +pub struct LegacyRecipient { + pub domain_idx: IDX, pub address: String, pub address_lcase: String, pub status: Status, HostResponse>, @@ -265,7 +420,16 @@ pub struct LegacyRecipient { pub orcpt: Option, } -#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)] +#[derive( + Debug, + Clone, + PartialEq, + Eq, + rkyv::Serialize, + rkyv::Deserialize, + rkyv::Archive, + serde::Deserialize, +)] pub struct LegacyDomain { pub domain: String, pub retry: Schedule, @@ -274,7 +438,16 @@ pub struct LegacyDomain { pub status: Status<(), LegacyError>, } -#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)] +#[derive( + Debug, + Clone, + PartialEq, + Eq, + rkyv::Serialize, + rkyv::Deserialize, + rkyv::Archive, + serde::Deserialize, +)] pub enum LegacyError { DnsError(String), UnexpectedResponse(HostResponse), @@ -287,7 +460,16 @@ pub enum LegacyError { Io(String), } -#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)] +#[derive( + Debug, + Clone, + PartialEq, + Eq, + rkyv::Serialize, + rkyv::Deserialize, + rkyv::Archive, + serde::Deserialize, +)] pub struct LegacyErrorDetails { pub entity: String, pub details: String, diff --git a/crates/migration/src/tasks.rs b/crates/migration/src/tasks.rs new file mode 100644 index 00000000..361ed8a8 --- /dev/null +++ b/crates/migration/src/tasks.rs @@ -0,0 +1,89 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use common::Server; +use store::{ + IterateParams, SUBSPACE_TASK_QUEUE, ValueKey, + write::{AnyClass, BatchBuilder, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now}, +}; +use trc::AddContext; +use utils::BlobHash; + +pub(crate) async fn migrate_tasks_v011(server: &Server) -> trc::Result<()> { + let from_key = ValueKey:: { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { + due: 0, + hash: BlobHash::default(), + }), + }; + let to_key = ValueKey:: { + account_id: u32::MAX, + collection: u8::MAX, + document_id: u32::MAX, + class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { + due: u64::MAX, + hash: BlobHash::default(), + }), + }; + + let now = now(); + let mut migrate_tasks = Vec::new(); + server + .core + .storage + .data + .iterate( + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + let due = key.deserialize_be_u64(0)?; + + if due > now { + migrate_tasks.push((key.to_vec(), value.to_vec())); + } + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + if !migrate_tasks.is_empty() { + let num_migrated = migrate_tasks.len(); + let mut batch = BatchBuilder::new(); + for (key, value) in migrate_tasks { + let mut new_key = key.clone(); + new_key[0..8].copy_from_slice(&now.to_be_bytes()); + + batch + .clear(ValueClass::Any(AnyClass { + subspace: SUBSPACE_TASK_QUEUE, + key, + })) + .set( + ValueClass::Any(AnyClass { + subspace: SUBSPACE_TASK_QUEUE, + key: new_key, + }), + value, + ); + } + server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + + trc::event!( + Server(trc::ServerEvent::Startup), + Details = format!("Migrated {num_migrated} tasks") + ); + } + + Ok(()) +} diff --git a/crates/nlp/Cargo.toml b/crates/nlp/Cargo.toml index ac9c67b0..e92da936 100644 --- a/crates/nlp/Cargo.toml +++ b/crates/nlp/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nlp" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/pop3/Cargo.toml b/crates/pop3/Cargo.toml index dcdc9e91..8cf26fee 100644 --- a/crates/pop3/Cargo.toml +++ b/crates/pop3/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "pop3" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/services/Cargo.toml b/crates/services/Cargo.toml index f56f5bf6..b5f71c0a 100644 --- a/crates/services/Cargo.toml +++ b/crates/services/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "services" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index 06b2e4b9..b0b997a3 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -200,6 +200,15 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver Session { // Queue message let source = if !self.is_authenticated() { - MessageSource::Unauthenticated + MessageSource::Unauthenticated( + dmarc_result.is_some_and(|result| result == DmarcResult::Pass), + ) } else { MessageSource::Authenticated }; - if self.is_authenticated() - || dmarc_result.is_some_and(|result| result == DmarcResult::Pass) - { - message.message.flags |= DMARC_AUTHENTICATED; - } if message .queue( Some(&headers), @@ -731,7 +727,6 @@ impl Session { let future_release = self.data.future_release; rcpt_to.sort_unstable(); for rcpt in rcpt_to { - let rcpt_idx = message.recipients.len(); message.recipients.push(queue::Recipient { address: rcpt.address, address_lcase: rcpt.address_lcase, @@ -754,7 +749,7 @@ impl Session { queue: QueueName::default(), }); - let envelope = QueueEnvelope::new_rcpt(&message, rcpt_idx); + let envelope = QueueEnvelope::new(&message, message.recipients.last().unwrap()); // Set next retry time let retry = if self.data.future_release == 0 { @@ -836,6 +831,7 @@ impl Session { MessageWrapper { queue_id, queue_name: QueueName::default(), + is_multi_queue: false, span_id, message, } diff --git a/crates/smtp/src/outbound/delivery.rs b/crates/smtp/src/outbound/delivery.rs index 66c46cfa..e088739b 100644 --- a/crates/smtp/src/outbound/delivery.rs +++ b/crates/smtp/src/outbound/delivery.rs @@ -15,7 +15,7 @@ use crate::outbound::mta_sts::lookup::MtaStsLookup; use crate::outbound::mta_sts::verify::VerifyPolicy; use crate::outbound::{client::StartTlsResult, dane::verify::TlsaVerify}; use crate::queue::dsn::SendDsn; -use crate::queue::spool::{LOCK_EXPIRY, SmtpSpool}; +use crate::queue::spool::SmtpSpool; use crate::queue::throttle::IsAllowed; use crate::queue::{ DomainPart, Error, FROM_REPORT, HostResponse, MessageWrapper, QueueEnvelope, QueuedMessage, @@ -33,7 +33,6 @@ use mail_auth::{ mta_sts::TlsRpt, report::tlsrpt::{FailureDetails, ResultType}, }; -use rand::Rng; use smtp_proto::MAIL_REQUIRETLS; use std::sync::Arc; use std::{ @@ -49,7 +48,7 @@ impl QueuedMessage { tokio::spawn(async move { // Lock queue event let queue_id = self.queue_id; - let status = if server.try_lock_event(queue_id).await { + let status = if server.try_lock_event(queue_id, self.queue_name).await { if let Some(mut message) = server.read_message(queue_id, self.queue_name).await { // Generate span id message.span_id = server.inner.data.span_id_gen.generate(); @@ -59,7 +58,7 @@ impl QueuedMessage { Delivery(DeliveryEvent::AttemptStart), SpanId = message.span_id, QueueId = message.queue_id, - QueueName = message.queue_name.as_str().to_string(), + QueueName = message.queue_name.to_string(), From = if !message.message.return_path.is_empty() { trc::Value::String(message.message.return_path.as_str().into()) } else { @@ -96,7 +95,7 @@ impl QueuedMessage { ); // Unlock event - server.unlock_event(queue_id).await; + server.unlock_event(queue_id, self.queue_name).await; queue_event } else { @@ -118,14 +117,12 @@ impl QueuedMessage { } // Unlock event - server.unlock_event(queue_id).await; + server.unlock_event(queue_id, self.queue_name).await; QueueEventStatus::Completed } } else { - QueueEventStatus::Locked { - until: now() + LOCK_EXPIRY + rand::rng().random_range(5..10), - } + QueueEventStatus::Locked }; // Notify queue manager @@ -133,7 +130,11 @@ impl QueuedMessage { .inner .ipc .queue_tx - .send(QueueEvent::WorkerDone { queue_id, status }) + .send(QueueEvent::WorkerDone { + queue_id, + queue_name: self.queue_name, + status, + }) .await .is_err() { @@ -185,23 +186,14 @@ impl QueuedMessage { .is_allowed(throttle, &message.message, message.span_id) .await { - // Save changes to disk - let now = now(); - let next_event = std::cmp::min( - retry_at, - message - .message - .next_event_after(self.queue_name.into(), now) - .unwrap_or(u64::MAX), - ); - trc::event!( Delivery(DeliveryEvent::RateLimitExceeded), Id = throttle.id.clone(), SpanId = span_id, - NextRetry = trc::Value::Timestamp(next_event) + NextRetry = trc::Value::Timestamp(retry_at) ); + let now = now(); for rcpt in message.message.recipients.iter_mut() { if matches!( &rcpt.status, @@ -234,7 +226,7 @@ impl QueuedMessage { ) && rcpt.retry.due <= now_ && rcpt.queue == message.queue_name { - let envelope = QueueEnvelope::new_rcpt(&message.message, rcpt_idx); + let envelope = QueueEnvelope::new(&message.message, rcpt); let gateway = server.get_gateway_or_default( &server .eval_if::(&queue_config.gateway, &envelope, message.span_id) @@ -260,7 +252,8 @@ impl QueuedMessage { ); // Build envelope - let mut envelope = QueueEnvelope::new_rcpt(&message.message, rcpt_idxs[0]); + let mut envelope = + QueueEnvelope::new(&message.message, &message.message.recipients[rcpt_idxs[0]]); // Throttle recipient domain for throttle in &queue_config.outbound_limiters.rcpt { @@ -1393,7 +1386,7 @@ impl MessageWrapper { self.message.recipients[rcpt_idx].status = status; if needs_retry { - let envelope = QueueEnvelope::new_rcpt(&self.message, rcpt_idx); + let envelope = QueueEnvelope::new(&self.message, &self.message.recipients[rcpt_idx]); let queue = server.get_queue_or_default( &server .eval_if::(&server.core.smtp.queue.queue, &envelope, self.span_id) diff --git a/crates/smtp/src/outbound/local.rs b/crates/smtp/src/outbound/local.rs index aa9ac761..919af0d2 100644 --- a/crates/smtp/src/outbound/local.rs +++ b/crates/smtp/src/outbound/local.rs @@ -7,8 +7,9 @@ use crate::{ outbound::DeliveryResult, queue::{ - DMARC_AUTHENTICATED, DomainPart, Error, ErrorDetails, HostResponse, MessageSource, - MessageWrapper, Status, UnexpectedResponse, quota::HasQueueQuota, spool::SmtpSpool, + DomainPart, Error, ErrorDetails, FROM_AUTHENTICATED, FROM_UNAUTHENTICATED_DMARC, + HostResponse, MessageSource, MessageWrapper, Status, UnexpectedResponse, + quota::HasQueueQuota, spool::SmtpSpool, }, reporting::SmtpReporting, }; @@ -37,7 +38,9 @@ impl MessageWrapper { let delivery_result = server .deliver_message(IngestMessage { sender_address: self.message.return_path_lcase.clone(), - sender_authenticated: self.message.flags & DMARC_AUTHENTICATED != 0, + sender_authenticated: self.message.flags + & (FROM_UNAUTHENTICATED_DMARC | FROM_AUTHENTICATED) + != 0, recipients: recipient_addresses, message_blob: self.message.blob_hash.clone(), message_size: self.message.size, diff --git a/crates/smtp/src/queue/dsn.rs b/crates/smtp/src/queue/dsn.rs index 7bb56283..2198974a 100644 --- a/crates/smtp/src/queue/dsn.rs +++ b/crates/smtp/src/queue/dsn.rs @@ -261,7 +261,7 @@ impl MessageWrapper { Status::TemporaryFailure(_) | Status::Scheduled ) && rcpt.notify.due <= now { - let envelope = QueueEnvelope::new_rcpt(&self.message, rcpt_idx); + let envelope = QueueEnvelope::new(&self.message, rcpt); let queue_id = server .eval_if::( diff --git a/crates/smtp/src/queue/manager.rs b/crates/smtp/src/queue/manager.rs index f9ca8a7f..45ca3555 100644 --- a/crates/smtp/src/queue/manager.rs +++ b/crates/smtp/src/queue/manager.rs @@ -4,11 +4,8 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{ - Message, QueueId, Status, - spool::{QUEUE_REFRESH, SmtpSpool}, -}; -use crate::queue::Recipient; +use super::{Message, QueueId, Status, spool::SmtpSpool}; +use crate::queue::{Recipient, spool::LOCK_EXPIRY}; use ahash::AHashMap; use common::{ Inner, @@ -16,7 +13,7 @@ use common::{ core::BuildServer, ipc::{QueueEvent, QueueEventStatus}, }; -use rand::seq::SliceRandom; +use rand::{Rng, seq::SliceRandom}; use std::{ collections::hash_map::Entry, sync::{Arc, atomic::Ordering}, @@ -27,15 +24,29 @@ use tokio::sync::mpsc; pub struct Queue { pub core: Arc, - pub on_hold: AHashMap, + pub locked_messages: LockedMessages, + pub stats: AHashMap, pub next_wake_up: Instant, pub rx: mpsc::Receiver, } #[derive(Debug)] -pub enum OnHold { - InFlight, - Locked { until: u64 }, +pub struct QueueStats { + pub in_flight: usize, + pub max_in_flight: usize, + pub last_warning: Instant, +} + +#[derive(Debug)] +pub struct LockedMessages { + pub locked: AHashMap<(QueueId, QueueName), LockedMessage>, + pub revision: u64, +} + +#[derive(Debug)] +pub struct LockedMessage { + pub expires: u64, + pub revision: u64, } impl SpawnQueue for mpsc::Receiver { @@ -46,14 +57,14 @@ impl SpawnQueue for mpsc::Receiver { } } -const CLEANUP_INTERVAL: Duration = Duration::from_secs(10 * 60); const BACK_PRESSURE_WARN_INTERVAL: Duration = Duration::from_secs(60); impl Queue { pub fn new(core: Arc, rx: mpsc::Receiver) -> Self { Queue { core, - on_hold: AHashMap::with_capacity(128), + locked_messages: LockedMessages::default(), + stats: AHashMap::new(), next_wake_up: Instant::now(), rx, } @@ -61,10 +72,6 @@ impl Queue { pub async fn start(&mut self) { let mut is_paused = false; - let mut next_cleanup = Instant::now() + CLEANUP_INTERVAL; - let mut last_backpressure_warning = Instant::now() - BACK_PRESSURE_WARN_INTERVAL; - let mut in_flight_count = 0; - let mut has_back_pressure = false; loop { let refresh_queue = match tokio::time::timeout( @@ -73,25 +80,37 @@ impl Queue { ) .await { - Ok(Some(QueueEvent::WorkerDone { queue_id, status })) => { - in_flight_count -= 1; + Ok(Some(QueueEvent::WorkerDone { + queue_id, + queue_name, + status, + })) => { + let queue_stats = self.stats.get_mut(&queue_name).unwrap(); + queue_stats.in_flight -= 1; match status { QueueEventStatus::Completed => { - self.on_hold.remove(&queue_id); - !self.on_hold.is_empty() || has_back_pressure + self.locked_messages.locked.remove(&(queue_id, queue_name)); + !self.locked_messages.locked.is_empty() || !queue_stats.has_capacity() } - QueueEventStatus::Locked { until } => { - let due_in = Instant::now() + Duration::from_secs(until - now()); + QueueEventStatus::Locked => { + let expires = LOCK_EXPIRY + rand::rng().random_range(5..10); + let due_in = Instant::now() + Duration::from_secs(expires); if due_in < self.next_wake_up { self.next_wake_up = due_in; } - self.on_hold.insert(queue_id, OnHold::Locked { until }); - self.on_hold.len() > 1 || has_back_pressure + self.locked_messages.locked.insert( + (queue_id, queue_name), + LockedMessage { + expires: now() + expires, + revision: self.locked_messages.revision, + }, + ); + self.locked_messages.locked.len() > 1 || !queue_stats.has_capacity() } QueueEventStatus::Deferred => { - self.on_hold.remove(&queue_id); + self.locked_messages.locked.remove(&(queue_id, queue_name)); true } } @@ -105,6 +124,18 @@ impl Queue { is_paused = paused; false } + Ok(Some(QueueEvent::ReloadSettings)) => { + let server = self.core.build_server(); + for (name, settings) in &server.core.smtp.queue.virtual_queues { + if let Some(stats) = self.stats.get_mut(name) { + stats.max_in_flight = settings.threads; + } else { + self.stats.insert(*name, QueueStats::new(settings.threads)); + } + } + + false + } Err(_) => true, Ok(Some(QueueEvent::Stop)) | Ok(None) => { break; @@ -114,127 +145,58 @@ impl Queue { if !is_paused { // Deliver scheduled messages if refresh_queue || self.next_wake_up <= Instant::now() { - // If the number of in-flight messages is greater than the maximum allowed, skip the queue - let server = self.core.build_server(); - let todo = "fix + implement virtual queues"; - let max_in_flight = 4; //server.core.smtp.queue.max_threads; - has_back_pressure = in_flight_count >= max_in_flight; - if has_back_pressure { - self.next_wake_up = Instant::now() + Duration::from_secs(QUEUE_REFRESH); - - if last_backpressure_warning.elapsed() >= BACK_PRESSURE_WARN_INTERVAL { - let queue_events = server.next_event().await; - last_backpressure_warning = Instant::now(); - trc::event!( - Queue(trc::QueueEvent::BackPressure), - Reason = - "Queue outbound processing capacity for this node exceeded.", - Total = queue_events.len(), - Details = self - .on_hold - .values() - .fold([0, 0], |mut acc, v| { - match v { - OnHold::InFlight => acc[0] += 1, - OnHold::Locked { .. } => acc[1] += 1, - } - acc - }) - .into_iter() - .map(trc::Value::from) - .collect::>(), - Limit = max_in_flight, - ); - } - continue; - } - // Process queue events - let now = now(); - let mut next_wake_up = QUEUE_REFRESH; - let mut queue_events = server.next_event().await; + let server = self.core.build_server(); + let mut queue_events = server.next_event(self).await; - if queue_events.len() > 5 { - queue_events.shuffle(&mut rand::rng()); + if queue_events.messages.len() > 3 { + queue_events.messages.shuffle(&mut rand::rng()); } - for queue_event in &queue_events { - if queue_event.due <= now { - // Enforce global concurrency limits - if in_flight_count >= max_in_flight { - has_back_pressure = true; - if last_backpressure_warning.elapsed() - >= BACK_PRESSURE_WARN_INTERVAL - { - last_backpressure_warning = Instant::now(); - trc::event!( - Queue(trc::QueueEvent::BackPressure), - Reason = "Queue outbound processing capacity for this node exceeded.", - Total = queue_events.len(), - Details = self - .on_hold - .values() - .fold([0, 0], |mut acc, v| { - match v { - OnHold::InFlight => acc[0] += 1, - OnHold::Locked { .. } => acc[1] += 1, - } - acc - }) - .into_iter() - .map(trc::Value::from) - .collect::>(), - Limit = max_in_flight, - ); - } - break; - } - - // Check if the message is still on hold - if let Some(on_hold) = self.on_hold.get(&queue_event.queue_id) { - match on_hold { - OnHold::Locked { until } => { - if *until > now { - let due_in = *until - now; - if due_in < next_wake_up { - next_wake_up = due_in; - } - continue; - } - } - OnHold::InFlight => continue, - } - - self.on_hold.remove(&queue_event.queue_id); + for queue_event in &queue_events.messages { + // Fetch queue stats + let stats = match self.stats.get_mut(&queue_event.queue_name) { + Some(stats) => stats, + None => { + let queue_config = + server.get_virtual_queue_or_default(&queue_event.queue_name); + self.stats.insert( + queue_event.queue_name, + QueueStats::new(queue_config.threads), + ); + self.stats.get_mut(&queue_event.queue_name).unwrap() } + }; + // Enforce concurrency limits + if stats.has_capacity() { // Deliver message - in_flight_count += 1; - self.on_hold.insert(queue_event.queue_id, OnHold::InFlight); + stats.in_flight += 1; queue_event.try_deliver(server.clone()); } else { - let due_in = queue_event.due - now; - if due_in < next_wake_up { - next_wake_up = due_in; + if stats.last_warning.elapsed() >= BACK_PRESSURE_WARN_INTERVAL { + stats.last_warning = Instant::now(); + trc::event!( + Queue(trc::QueueEvent::BackPressure), + Reason = "Processing capacity for this queue exceeded.", + QueueName = queue_event.queue_name.to_string(), + Limit = stats.max_in_flight, + ); } + self.locked_messages + .locked + .remove(&(queue_event.queue_id, queue_event.queue_name)); } } // Remove expired locks - let now = Instant::now(); - if next_cleanup <= now { - next_cleanup = now + CLEANUP_INTERVAL; + let now = now(); + self.locked_messages.locked.retain(|_, locked| { + locked.expires > now && locked.revision == self.locked_messages.revision + }); - if !self.on_hold.is_empty() { - let now = store::write::now(); - self.on_hold.retain(|queue_id, status| match status { - OnHold::InFlight => true, - OnHold::Locked { until } => *until > now, - }); - } - } - - self.next_wake_up = now + Duration::from_secs(next_wake_up); + self.next_wake_up = Instant::now() + + Duration::from_secs(queue_events.next_refresh.saturating_sub(now)); } } else { // Queue is paused @@ -330,34 +292,6 @@ impl Message { expires } - pub fn next_event_after(&self, queue: Option, instant: u64) -> Option { - let mut next_event = None; - - for rcpt in &self.recipients { - if matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_)) - && queue.is_none_or(|q| rcpt.queue == q) - { - if rcpt.retry.due > instant - && next_event.as_ref().is_none_or(|ne| rcpt.retry.due.lt(ne)) - { - next_event = rcpt.retry.due.into(); - } - if rcpt.notify.due > instant - && next_event.as_ref().is_none_or(|ne| rcpt.notify.due.lt(ne)) - { - next_event = rcpt.notify.due.into(); - } - if let Some(expires) = rcpt.expiration_time(self.created) { - if expires > instant && next_event.as_ref().is_none_or(|ne| expires.lt(ne)) { - next_event = expires.into(); - } - } - } - } - - next_event - } - pub fn next_events(&self) -> AHashMap { let mut next_events = AHashMap::new(); @@ -406,3 +340,27 @@ impl Recipient { pub trait SpawnQueue { fn spawn(self, core: Arc); } + +impl QueueStats { + fn new(max_in_flight: usize) -> Self { + QueueStats { + in_flight: 0, + max_in_flight, + last_warning: Instant::now() - BACK_PRESSURE_WARN_INTERVAL, + } + } + + #[inline] + pub fn has_capacity(&self) -> bool { + self.in_flight < self.max_in_flight + } +} + +impl Default for LockedMessages { + fn default() -> Self { + LockedMessages { + locked: AHashMap::with_capacity(128), + revision: 0, + } + } +} diff --git a/crates/smtp/src/queue/mod.rs b/crates/smtp/src/queue/mod.rs index 03ef0c07..67a5d737 100644 --- a/crates/smtp/src/queue/mod.rs +++ b/crates/smtp/src/queue/mod.rs @@ -42,7 +42,7 @@ pub struct QueuedMessage { #[derive(Debug, Clone, Copy)] pub enum MessageSource { Authenticated, - Unauthenticated, + Unauthenticated(bool), Dsn, Report, Autogenerated, @@ -73,6 +73,7 @@ pub struct Message { pub struct MessageWrapper { pub queue_id: QueueId, pub queue_name: QueueName, + pub is_multi_queue: bool, pub span_id: u64, pub message: Message, } @@ -116,11 +117,15 @@ pub struct Recipient { pub orcpt: Option, } -pub const FROM_REPORT: u64 = 1 << 32; -pub const DMARC_AUTHENTICATED: u64 = 2 << 32; +pub const FROM_AUTHENTICATED: u64 = 1 << 32; +pub const FROM_UNAUTHENTICATED: u64 = 1 << 33; +pub const FROM_UNAUTHENTICATED_DMARC: u64 = 1 << 34; +pub const FROM_DSN: u64 = 1 << 35; +pub const FROM_REPORT: u64 = 1 << 36; +pub const FROM_AUTOGENERATED: u64 = 1 << 37; pub const RCPT_DSN_SENT: u64 = 1 << 32; -pub const RCPT_STATUS_CHANGED: u64 = 2 << 32; +pub const RCPT_STATUS_CHANGED: u64 = 1 << 33; #[derive( Debug, @@ -252,26 +257,24 @@ impl Schedule { pub struct QueueEnvelope<'x> { pub message: &'x Message, + pub domain: &'x str, pub mx: &'x str, + pub rcpt: &'x Recipient, pub remote_ip: IpAddr, pub local_ip: IpAddr, - pub current_rcpt: usize, } impl<'x> QueueEnvelope<'x> { - pub fn new_rcpt(message: &'x Message, current_rcpt: usize) -> Self { + pub fn new(message: &'x Message, rcpt: &'x Recipient) -> Self { Self { message, - current_rcpt, + domain: rcpt.address_lcase.domain_part(), + rcpt, mx: "", remote_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), local_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), } } - - fn current_recipient(&self) -> Option<&'x Recipient> { - self.message.recipients.get(self.current_rcpt) - } } impl<'x> ResolveVariable for QueueEnvelope<'x> { @@ -279,16 +282,8 @@ impl<'x> ResolveVariable for QueueEnvelope<'x> { match variable { V_SENDER => self.message.return_path_lcase.as_str().into(), V_SENDER_DOMAIN => self.message.return_path_domain.as_str().into(), - V_RECIPIENT_DOMAIN => self - .current_recipient() - .map(|d| d.address_lcase.domain_part()) - .unwrap_or_default() - .into(), - V_RECIPIENT => self - .current_recipient() - .map(|r| r.address_lcase.as_str()) - .unwrap_or_default() - .into(), + V_RECIPIENT_DOMAIN => self.domain.into(), + V_RECIPIENT => self.rcpt.address_lcase.as_str().into(), V_RECIPIENTS => self .message .recipients @@ -296,55 +291,56 @@ impl<'x> ResolveVariable for QueueEnvelope<'x> { .map(|r| Variable::from(r.address_lcase.as_str())) .collect::>() .into(), - V_QUEUE_RETRY_NUM => self - .current_recipient() - .map(|d| d.retry.inner) - .unwrap_or_default() - .into(), - V_QUEUE_NOTIFY_NUM => self - .current_recipient() - .map(|d| d.notify.inner) - .unwrap_or_default() - .into(), - V_QUEUE_EXPIRES_IN => self - .current_recipient() - .map(|d| match &d.expires { - QueueExpiry::Duration(time) => { - (*time + self.message.created).saturating_sub(now()) + V_QUEUE_RETRY_NUM => self.rcpt.retry.inner.into(), + V_QUEUE_NOTIFY_NUM => self.rcpt.notify.inner.into(), + V_QUEUE_EXPIRES_IN => match &self.rcpt.expires { + QueueExpiry::Duration(time) => (*time + self.message.created).saturating_sub(now()), + QueueExpiry::Count(count) => (*count) as u64, + } + .into(), + V_QUEUE_LAST_STATUS => self.rcpt.status.to_compact_string().into(), + V_QUEUE_LAST_ERROR => match &self.rcpt.status { + Status::Scheduled | Status::Completed(_) => "none", + Status::TemporaryFailure(err) | Status::PermanentFailure(err) => { + match &err.details { + Error::DnsError(_) => "dns", + Error::UnexpectedResponse(_) => "unexpected-reply", + Error::ConnectionError(_) => "connection", + Error::TlsError(_) => "tls", + Error::DaneError(_) => "dane", + Error::MtaStsError(_) => "mta-sts", + Error::RateLimited => "rate", + Error::ConcurrencyLimited => "concurrency", + Error::Io(_) => "io", } - QueueExpiry::Count(count) => (*count) as u64, - }) - .unwrap_or_default() - .into(), - V_QUEUE_LAST_STATUS => self - .current_recipient() - .map(|d| d.status.to_compact_string()) - .unwrap_or_default() - .into(), - V_QUEUE_LAST_ERROR => self - .current_recipient() - .map(|d| match &d.status { - Status::Scheduled | Status::Completed(_) => "none", - Status::TemporaryFailure(err) | Status::PermanentFailure(err) => { - match &err.details { - Error::DnsError(_) => "dns", - Error::UnexpectedResponse(_) => "unexpected-reply", - Error::ConnectionError(_) => "connection", - Error::TlsError(_) => "tls", - Error::DaneError(_) => "dane", - Error::MtaStsError(_) => "mta-sts", - Error::RateLimited => "rate", - Error::ConcurrencyLimited => "concurrency", - Error::Io(_) => "io", - } - } - }) - .unwrap_or_default() - .into(), + } + } + .into(), + V_QUEUE_NAME => self.rcpt.queue.as_str().into(), + V_QUEUE_AGE => now().saturating_sub(self.message.created).into(), + V_SOURCE => if (self.message.flags & FROM_AUTHENTICATED) != 0 { + "authenticated" + } else if (self.message.flags & FROM_UNAUTHENTICATED_DMARC) != 0 { + "dmarc_pass" + } else if (self.message.flags & FROM_UNAUTHENTICATED) != 0 { + "unauthenticated" + } else if (self.message.flags & FROM_DSN) != 0 { + "dsn" + } else if (self.message.flags & FROM_REPORT) != 0 { + "report" + } else if (self.message.flags & FROM_AUTOGENERATED) != 0 { + "autogenerated" + } else { + "unknown" + } + .into(), V_MX => self.mx.into(), V_PRIORITY => self.message.priority.into(), V_REMOTE_IP => self.remote_ip.to_compact_string().into(), V_LOCAL_IP => self.local_ip.to_compact_string().into(), + V_RECEIVED_FROM_IP => self.message.received_from_ip.to_compact_string().into(), + V_RECEIVED_VIA_PORT => self.message.received_via_port.into(), + V_SIZE => self.message.size.into(), _ => "".into(), } } diff --git a/crates/smtp/src/queue/quota.rs b/crates/smtp/src/queue/quota.rs index 428dd86c..5363f40e 100644 --- a/crates/smtp/src/queue/quota.rs +++ b/crates/smtp/src/queue/quota.rs @@ -68,7 +68,7 @@ impl HasQueueQuota for Server { && !self .check_quota( quota, - &QueueEnvelope::new_rcpt(&message.message, rcpt_idx), + &QueueEnvelope::new(&message.message, rcpt), message.message.size, ((rcpt_idx + 1) << 32) as u64, &mut quota_keys, @@ -90,11 +90,11 @@ impl HasQueueQuota for Server { } for quota in &self.core.smtp.queue.quota.rcpt { - for rcpt_idx in 0..message.message.recipients.len() { + for (rcpt_idx, rcpt) in message.message.recipients.iter().enumerate() { if !self .check_quota( quota, - &QueueEnvelope::new_rcpt(&message.message, rcpt_idx), + &QueueEnvelope::new(&message.message, rcpt), message.message.size, (rcpt_idx + 1) as u64, &mut quota_keys, diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index 08ef99e2..384a0984 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -8,11 +8,16 @@ use super::{ ArchivedMessage, ArchivedStatus, Message, MessageSource, QueueEnvelope, QueueId, QueuedMessage, QuotaKey, Recipient, Schedule, Status, }; -use crate::queue::{DomainPart, MessageWrapper}; +use crate::queue::manager::{LockedMessage, Queue}; +use crate::queue::{ + DomainPart, FROM_AUTHENTICATED, FROM_AUTOGENERATED, FROM_DSN, FROM_REPORT, + FROM_UNAUTHENTICATED, FROM_UNAUTHENTICATED_DMARC, MessageWrapper, +}; use common::config::smtp::queue::{QueueExpiry, QueueName}; use common::ipc::QueueEvent; use common::{KV_LOCK_QUEUE_MESSAGE, Server}; use std::borrow::Cow; +use std::collections::hash_map::Entry; use std::future::Future; use std::net::{IpAddr, Ipv4Addr}; use std::time::SystemTime; @@ -20,12 +25,18 @@ use store::write::key::DeserializeBigEndian; use store::write::{ AlignedBytes, Archive, Archiver, BatchBuilder, BlobOp, QueueClass, ValueClass, now, }; -use store::{IterateParams, Serialize, SerializeInfallible, U64_LEN, ValueKey}; -use trc::ServerEvent; +use store::{Deserialize, IterateParams, Serialize, SerializeInfallible, U64_LEN, ValueKey}; +use trc::{AddContext, ServerEvent}; use utils::BlobHash; -pub const LOCK_EXPIRY: u64 = 300; -pub const QUEUE_REFRESH: u64 = 300; +pub const LOCK_EXPIRY: u64 = 10 * 60; // 10 minutes +pub const QUEUE_REFRESH: u64 = 5 * 60; // 5 minutes +const INFINITE_LOCK: u64 = 60 * 60 * 24 * 365; // 1 year + +pub struct QueuedMessages { + pub messages: Vec, + pub next_refresh: u64, +} pub trait SmtpSpool: Sync + Send { fn new_message( @@ -36,11 +47,19 @@ pub trait SmtpSpool: Sync + Send { span_id: u64, ) -> MessageWrapper; - fn next_event(&self) -> impl Future> + Send; + fn next_event(&self, queue: &mut Queue) -> impl Future + Send; - fn try_lock_event(&self, queue_id: QueueId) -> impl Future + Send; + fn try_lock_event( + &self, + queue_id: QueueId, + queue_name: QueueName, + ) -> impl Future + Send; - fn unlock_event(&self, queue_id: QueueId) -> impl Future + Send; + fn unlock_event( + &self, + queue_id: QueueId, + queue_name: QueueName, + ) -> impl Future + Send; fn read_message( &self, @@ -69,6 +88,7 @@ impl SmtpSpool for Server { MessageWrapper { queue_id: self.inner.data.queue_id_gen.generate(), queue_name: QueueName::default(), + is_multi_queue: false, span_id, message: Message { created, @@ -88,7 +108,7 @@ impl SmtpSpool for Server { } } - async fn next_event(&self) -> Vec { + async fn next_event(&self, queue: &mut Queue) -> QueuedMessages { let now = now(); let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent( store::write::QueueEvent { @@ -105,26 +125,69 @@ impl SmtpSpool for Server { }, ))); - let mut events = Vec::new(); + let mut events = QueuedMessages { + messages: Vec::new(), + next_refresh: now + QUEUE_REFRESH, + }; + queue.locked_messages.revision += 1; let result = self .store() .iterate( IterateParams::new(from_key, to_key).ascending().no_values(), |key, _| { let due = key.deserialize_be_u64(0)?; - let queue_id = key.deserialize_be_u64(U64_LEN)?; - let queue_name = - QueueName::from_bytes(key.get(U64_LEN + U64_LEN..).unwrap_or_default()) - .unwrap_or_default(); - events.push(QueuedMessage { - due, - queue_id, - queue_name, - }); + if due <= now { + let queue_id = key.deserialize_be_u64(U64_LEN)?; + let queue_name = + QueueName::from_bytes(key.get(U64_LEN + U64_LEN..).unwrap_or_default()) + .unwrap_or_default(); - Ok(due <= now) + let add_event = queue + .stats + .get(&queue_name) + .is_none_or(|stats| stats.has_capacity()) + && match queue.locked_messages.locked.entry((queue_id, queue_name)) { + Entry::Occupied(mut entry) => { + let locked = entry.get_mut(); + locked.revision = queue.locked_messages.revision; + if locked.expires <= now { + locked.expires = now + INFINITE_LOCK; + + true + } else { + if locked.expires < events.next_refresh { + events.next_refresh = locked.expires; + } + + false + } + } + Entry::Vacant(entry) => { + entry.insert(LockedMessage { + expires: now + INFINITE_LOCK, + revision: queue.locked_messages.revision, + }); + true + } + }; + + if add_event { + events.messages.push(QueuedMessage { + due, + queue_id, + queue_name, + }); + } + + Ok(true) + } else { + if due < events.next_refresh { + events.next_refresh = due; + } + Ok(false) + } }, ) .await; @@ -139,15 +202,23 @@ impl SmtpSpool for Server { events } - async fn try_lock_event(&self, queue_id: QueueId) -> bool { + async fn try_lock_event(&self, queue_id: QueueId, queue_name: QueueName) -> bool { match self .in_memory_store() - .try_lock(KV_LOCK_QUEUE_MESSAGE, &queue_id.to_be_bytes(), LOCK_EXPIRY) + .try_lock( + KV_LOCK_QUEUE_MESSAGE, + &lock_id(queue_id, queue_name), + LOCK_EXPIRY, + ) .await { Ok(result) => { if !result { - trc::event!(Queue(trc::QueueEvent::Locked), QueueId = queue_id,); + trc::event!( + Queue(trc::QueueEvent::Locked), + QueueId = queue_id, + QueueName = queue_name.to_string() + ); } result } @@ -161,10 +232,10 @@ impl SmtpSpool for Server { } } - async fn unlock_event(&self, queue_id: QueueId) { + async fn unlock_event(&self, queue_id: QueueId, queue_name: QueueName) { if let Err(err) = self .in_memory_store() - .remove_lock(KV_LOCK_QUEUE_MESSAGE, &queue_id.to_be_bytes()) + .remove_lock(KV_LOCK_QUEUE_MESSAGE, &lock_id(queue_id, queue_name)) .await { trc::error!( @@ -187,6 +258,10 @@ impl SmtpSpool for Server { None => Ok(None), }) { Ok(Some(message)) => Some(MessageWrapper { + is_multi_queue: message.recipients.iter().any(|rcpt| { + matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_)) + && rcpt.queue != queue_name + }), queue_id, queue_name, span_id: 0, @@ -216,6 +291,13 @@ impl SmtpSpool for Server { } } +fn lock_id(queue_id: QueueId, queue_name: QueueName) -> [u8; 16] { + let mut id = [0; 16]; + id[..8].copy_from_slice(&queue_id.to_be_bytes()); + id[8..].copy_from_slice(queue_name.as_ref()); + id +} + impl MessageWrapper { pub async fn queue( mut self, @@ -225,6 +307,26 @@ impl MessageWrapper { server: &Server, source: MessageSource, ) -> bool { + // Set flags + let (flags, event) = match source { + MessageSource::Authenticated => ( + FROM_AUTHENTICATED, + trc::QueueEvent::QueueMessageAuthenticated, + ), + MessageSource::Unauthenticated(true) => { + (FROM_UNAUTHENTICATED_DMARC, trc::QueueEvent::QueueMessage) + } + MessageSource::Unauthenticated(false) => { + (FROM_UNAUTHENTICATED, trc::QueueEvent::QueueMessage) + } + MessageSource::Dsn => (FROM_DSN, trc::QueueEvent::QueueDsn), + MessageSource::Report => (FROM_REPORT, trc::QueueEvent::QueueReport), + MessageSource::Autogenerated => { + (FROM_AUTOGENERATED, trc::QueueEvent::QueueAutogenerated) + } + }; + self.message.flags |= flags; + // Write blob let message = if let Some(raw_headers) = raw_headers { let mut message = Vec::with_capacity(raw_headers.len() + raw_message.len()); @@ -275,13 +377,7 @@ impl MessageWrapper { } trc::event!( - Queue(match source { - MessageSource::Authenticated => trc::QueueEvent::QueueMessageAuthenticated, - MessageSource::Unauthenticated => trc::QueueEvent::QueueMessage, - MessageSource::Dsn => trc::QueueEvent::QueueDsn, - MessageSource::Report => trc::QueueEvent::QueueReport, - MessageSource::Autogenerated => trc::QueueEvent::QueueAutogenerated, - }), + Queue(event), SpanId = session_id, QueueId = self.queue_id, From = if !self.message.return_path.is_empty() { @@ -403,7 +499,6 @@ impl MessageWrapper { server: &Server, ) { // Resolve queue - let idx = self.message.recipients.len(); self.message.recipients.push(Recipient { address: rcpt.into(), address_lcase: rcpt_lcase.into(), @@ -419,7 +514,7 @@ impl MessageWrapper { &server .eval_if::( &server.core.smtp.queue.queue, - &QueueEnvelope::new_rcpt(&self.message, idx), + &QueueEnvelope::new(&self.message, self.message.recipients.last().unwrap()), self.span_id, ) .await @@ -467,26 +562,68 @@ impl MessageWrapper { ); } - let span_id = self.span_id; - batch.set( - ValueClass::Queue(QueueClass::Message(self.queue_id)), - match Archiver::new(self.message).serialize() { - Ok(data) => data, - Err(err) => { - trc::error!( - err.details("Failed to serialize message.") - .span_id(span_id) + if self.is_multi_queue { + batch.merge( + ValueClass::Queue(QueueClass::Message(self.queue_id)), + move |bytes| { + let mut cur_message = as Deserialize>::deserialize( + bytes.ok_or_else(|| { + trc::StoreEvent::NotFound + .into_err() + .details("Message no longer exists.") + .caused_by(trc::location!()) + .ctx(trc::Key::QueueId, self.queue_id) + })?, + ) + .and_then(|archive| archive.deserialize::()) + .caused_by(trc::location!())?; + + if cur_message.blob_hash == self.message.blob_hash + && cur_message.recipients.len() == self.message.recipients.len() + { + for (rcpt_idx, rcpt) in self + .message + .recipients + .iter() + .enumerate() + .filter(|(_, rcpt)| rcpt.queue == self.queue_name) + { + cur_message.recipients[rcpt_idx] = rcpt.clone(); + } + + Archiver::new(cur_message) + .serialize() .caused_by(trc::location!()) - ); - return false; - } - }, - ); + } else { + Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("Message blob hash or recipient count mismatch.") + .caused_by(trc::location!()) + .ctx(trc::Key::QueueId, self.queue_id)) + } + }, + ); + } else { + batch.set( + ValueClass::Queue(QueueClass::Message(self.queue_id)), + match Archiver::new(self.message).serialize() { + Ok(data) => data, + Err(err) => { + trc::error!( + err.details("Failed to serialize message.") + .span_id(self.span_id) + .caused_by(trc::location!()) + ); + return false; + } + }, + ); + } if let Err(err) = server.store().write(batch.build_all()).await { trc::error!( err.details("Failed to save changes.") - .span_id(span_id) + .span_id(self.span_id) .caused_by(trc::location!()) ); false diff --git a/crates/spam-filter/Cargo.toml b/crates/spam-filter/Cargo.toml index 00a5dab2..ef28146f 100644 --- a/crates/spam-filter/Cargo.toml +++ b/crates/spam-filter/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "spam-filter" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index e6d96b0e..5ec0e672 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "store" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/store/src/backend/foundationdb/write.rs b/crates/store/src/backend/foundationdb/write.rs index 092a96f6..c9fd3850 100644 --- a/crates/store/src/backend/foundationdb/write.rs +++ b/crates/store/src/backend/foundationdb/write.rs @@ -27,6 +27,7 @@ use std::{ cmp::Ordering, time::{Duration, Instant}, }; +use trc::AddContext; impl FdbStore { pub(crate) async fn write(&self, batch: Batch<'_>) -> trc::Result { @@ -133,6 +134,21 @@ impl FdbStore { trx.set(&key, &num.to_le_bytes()[..]); result.push_counter_id(num); } + ValueOp::Merge(merge) => { + let value = match read_chunked_value(&key, &trx, false) + .await + .caused_by(trc::location!())? + { + ChunkedValue::Single(slice) => { + (merge.fnc)(Some(slice.as_ref())) + } + ChunkedValue::Chunked { bytes, .. } => { + (merge.fnc)(Some(bytes.as_ref())) + } + ChunkedValue::None => (merge.fnc)(None), + }?; + trx.set(&key, value.as_ref()); + } ValueOp::Clear => { if do_chunk { trx.clear_range( diff --git a/crates/store/src/backend/mysql/write.rs b/crates/store/src/backend/mysql/write.rs index 9bf7ccfb..fe20c827 100644 --- a/crates/store/src/backend/mysql/write.rs +++ b/crates/store/src/backend/mysql/write.rs @@ -4,12 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::time::{Duration, Instant}; - -use ahash::AHashMap; -use mysql_async::{Conn, Error, IsolationLevel, TxOpts, params, prelude::Queryable}; -use rand::Rng; - +use super::{MysqlStore, into_error}; use crate::{ IndexKey, Key, LogKey, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER, SUBSPACE_QUOTA, U64_LEN, write::{ @@ -17,8 +12,10 @@ use crate::{ ValueClass, ValueOp, }, }; - -use super::{MysqlStore, into_error}; +use ahash::AHashMap; +use mysql_async::{Conn, Error, IsolationLevel, TxOpts, params, prelude::Queryable}; +use rand::Rng; +use std::time::{Duration, Instant}; #[derive(Debug)] enum CommitError { @@ -221,6 +218,39 @@ impl MysqlStore { })?, ); } + ValueOp::Merge(merge) => { + let s = trx + .prep(format!("SELECT v FROM {} WHERE k = ? FOR UPDATE", table)) + .await?; + let (exists, value) = trx + .exec_first::, _, _>(&s, (&key,)) + .await? + .map(|bytes| { + (merge.fnc)(Some(bytes.as_ref())) + .map(|v| (true, v)) + .map_err(CommitError::from) + }) + .unwrap_or_else(|| { + (merge.fnc)(None) + .map(|v| (false, v)) + .map_err(CommitError::from) + })?; + + let s = if exists { + trx.prep(format!("UPDATE {} SET v = :v WHERE k = :k", table)) + .await? + } else { + trx.prep(format!("INSERT INTO {} (k, v) VALUES (:k, :v)", table)) + .await? + }; + + if let Err(err) = + trx.exec_drop(&s, params! {"k" => key, "v" => &value}).await + { + trx.rollback().await?; + return Err(err.into()); + } + } ValueOp::Clear => { // Update asserted value if let Some(exists) = asserted_values.get_mut(&key) { diff --git a/crates/store/src/backend/postgres/write.rs b/crates/store/src/backend/postgres/write.rs index 5c56a978..4faa07a9 100644 --- a/crates/store/src/backend/postgres/write.rs +++ b/crates/store/src/backend/postgres/write.rs @@ -4,13 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::time::{Duration, Instant}; - -use ahash::AHashMap; -use deadpool_postgres::Object; -use rand::Rng; -use tokio_postgres::{IsolationLevel, error::SqlState}; - +use super::{PostgresStore, into_error}; use crate::{ IndexKey, Key, LogKey, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER, SUBSPACE_QUOTA, U64_LEN, write::{ @@ -18,8 +12,11 @@ use crate::{ ValueClass, ValueOp, }, }; - -use super::{PostgresStore, into_error}; +use ahash::AHashMap; +use deadpool_postgres::Object; +use rand::Rng; +use std::time::{Duration, Instant}; +use tokio_postgres::{IsolationLevel, error::SqlState}; #[derive(Debug)] enum CommitError { @@ -212,6 +209,47 @@ impl PostgresStore { .and_then(|row| row.try_get::<_, i64>(0))?, ); } + ValueOp::Merge(merge) => { + let s = trx + .prepare_cached(&format!( + "SELECT v FROM {} WHERE k = $1 FOR UPDATE", + table + )) + .await?; + let (exists, value) = trx + .query_opt(&s, &[&key]) + .await? + .map(|row| { + row.try_get::<_, &[u8]>(0) + .map_err(CommitError::from) + .and_then(|v| { + (merge.fnc)(Some(v)) + .map(|v| (true, v)) + .map_err(CommitError::from) + }) + }) + .unwrap_or_else(|| { + (merge.fnc)(None) + .map(|v| (false, v)) + .map_err(CommitError::from) + })?; + + let s = if exists { + trx.prepare_cached(&format!( + "UPDATE {} SET v = $2 WHERE k = $1", + table + )) + .await? + } else { + trx.prepare_cached(&format!( + "INSERT INTO {} (k, v) VALUES ($1, $2)", + table + )) + .await? + }; + + trx.execute(&s, &[&key, &value]).await?; + } ValueOp::Clear => { let s = trx .prepare_cached(&format!("DELETE FROM {} WHERE k = $1", table)) diff --git a/crates/store/src/backend/rocksdb/write.rs b/crates/store/src/backend/rocksdb/write.rs index 9be0d4be..8b7d0638 100644 --- a/crates/store/src/backend/rocksdb/write.rs +++ b/crates/store/src/backend/rocksdb/write.rs @@ -4,18 +4,6 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{ - sync::Arc, - thread::sleep, - time::{Duration, Instant}, -}; - -use rand::Rng; -use rocksdb::{ - BoundColumnFamily, ErrorKind, IteratorMode, OptimisticTransactionDB, - OptimisticTransactionOptions, WriteOptions, -}; - use super::{CF_INDEXES, CF_LOGS, CfHandle, RocksDbStore, into_error}; use crate::{ Deserialize, IndexKey, Key, LogKey, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER, @@ -25,6 +13,16 @@ use crate::{ AssignedIds, Batch, MAX_COMMIT_ATTEMPTS, MAX_COMMIT_TIME, Operation, ValueClass, ValueOp, }, }; +use rand::Rng; +use rocksdb::{ + BoundColumnFamily, ErrorKind, IteratorMode, OptimisticTransactionDB, + OptimisticTransactionOptions, WriteOptions, +}; +use std::{ + sync::Arc, + thread::sleep, + time::{Duration, Instant}, +}; impl RocksDbStore { pub(crate) async fn write(&self, mut batch: Batch<'_>) -> trc::Result { @@ -223,6 +221,12 @@ impl RocksDBTransaction<'_, '_> { txn.put_cf(&cf, &key, &num.to_le_bytes()[..])?; result.push_counter_id(num); } + ValueOp::Merge(merge) => { + let value = (merge.fnc)( + txn.get_pinned_for_update_cf(&cf, &key, true)?.as_deref(), + )?; + txn.put_cf(&cf, &key, value)?; + } ValueOp::Clear => { txn.delete_cf(&cf, &key)?; } diff --git a/crates/store/src/backend/sqlite/write.rs b/crates/store/src/backend/sqlite/write.rs index c70e21f1..4fea537b 100644 --- a/crates/store/src/backend/sqlite/write.rs +++ b/crates/store/src/backend/sqlite/write.rs @@ -135,6 +135,29 @@ impl SqliteStore { .caused_by(trc::location!())?, ); } + ValueOp::Merge(merge) => { + let value = trx + .prepare_cached(&format!("SELECT v FROM {} WHERE k = ?", table)) + .map_err(into_error) + .caused_by(trc::location!())? + .query_row([&key], |row| { + Ok((merge.fnc)(Some(row.get_ref(0)?.as_bytes()?))) + }) + .optional() + .map_err(into_error) + .caused_by(trc::location!())? + .unwrap_or_else(|| (merge.fnc)(None))?; + + trx.prepare_cached(&format!( + "INSERT OR REPLACE INTO {} (k, v) VALUES (?, ?)", + table + )) + .map_err(into_error) + .caused_by(trc::location!())? + .execute([&key, &value]) + .map_err(into_error) + .caused_by(trc::location!())?; + } ValueOp::Clear => { trx.prepare_cached(&format!("DELETE FROM {} WHERE k = ?", table)) .map_err(into_error) diff --git a/crates/store/src/write/batch.rs b/crates/store/src/write/batch.rs index 6352fc92..f52b17aa 100644 --- a/crates/store/src/write/batch.rs +++ b/crates/store/src/write/batch.rs @@ -8,7 +8,7 @@ use super::{ Batch, BatchBuilder, BitmapClass, ChangedCollection, IntoOperations, Operation, TagValue, ValueClass, ValueOp, assert::ToAssertValue, log::VanishedItem, }; -use crate::{SerializeInfallible, U32_LEN}; +use crate::{SerializeInfallible, U32_LEN, write::MergeFn}; use utils::map::{bitmap::ShortId, vec_map::VecMap}; impl BatchBuilder { @@ -192,6 +192,21 @@ impl BatchBuilder { self } + pub fn merge( + &mut self, + class: impl Into, + value: impl Fn(Option<&[u8]>) -> trc::Result> + Sync + Send + 'static, + ) -> &mut Self { + self.ops.push(Operation::Value { + class: class.into(), + op: ValueOp::Merge(MergeFn { + fnc: Box::new(value), + fnc_id: rand::random::(), + }), + }); + self + } + pub fn set_versioned( &mut self, class: impl Into, diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 1bab5afa..b8600599 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -296,10 +296,17 @@ pub enum ValueOp { }, AtomicAdd(i64), AddAndGet(i64), + Merge(MergeFn), #[default] Clear, } +#[allow(clippy::type_complexity)] +pub struct MergeFn { + pub fnc: Box) -> trc::Result> + Send + Sync>, + pub fnc_id: u64, +} + #[derive(Debug, PartialEq, Clone, Eq, Hash)] pub enum BlobOp { Reserve { hash: BlobHash, until: u64 }, @@ -507,3 +514,25 @@ impl ArchiveVersion { } } } + +impl std::fmt::Debug for MergeFn { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MergeFn") + .field("fnc_id", &self.fnc_id) + .finish() + } +} + +impl PartialEq for MergeFn { + fn eq(&self, other: &Self) -> bool { + self.fnc_id == other.fnc_id + } +} + +impl Eq for MergeFn {} + +impl std::hash::Hash for MergeFn { + fn hash(&self, state: &mut H) { + self.fnc_id.hash(state); + } +} diff --git a/crates/trc/Cargo.toml b/crates/trc/Cargo.toml index c095c749..44e58647 100644 --- a/crates/trc/Cargo.toml +++ b/crates/trc/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "trc" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/trc/event-macro/Cargo.toml b/crates/trc/event-macro/Cargo.toml index 84d35546..f8ee8fcb 100644 --- a/crates/trc/event-macro/Cargo.toml +++ b/crates/trc/event-macro/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "event_macro" -version = "0.12.5" +version = "0.13.0" edition = "2024" [lib] diff --git a/crates/utils/Cargo.toml b/crates/utils/Cargo.toml index c969915d..c9b204d7 100644 --- a/crates/utils/Cargo.toml +++ b/crates/utils/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "utils" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/crates/utils/proc-macros/Cargo.toml b/crates/utils/proc-macros/Cargo.toml index 1ac3bc9b..2548ff5f 100644 --- a/crates/utils/proc-macros/Cargo.toml +++ b/crates/utils/proc-macros/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "proc_macros" -version = "0.12.5" +version = "0.13.0" edition = "2024" [lib] diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 3a633f05..81316422 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tests" -version = "0.12.5" +version = "0.13.0" edition = "2024" resolver = "2" diff --git a/tests/src/jmap/auth_limits.rs b/tests/src/jmap/auth_limits.rs index bfa20538..ae98b4a0 100644 --- a/tests/src/jmap/auth_limits.rs +++ b/tests/src/jmap/auth_limits.rs @@ -55,7 +55,7 @@ pub async fn test(params: &mut JMAPTest) { assert!(matches!( Client::new() .credentials(Credentials::basic("jdoe@example.com", "abcde")) - .accept_invalid_certs(true) + .accept_invalid_certs(true) .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await, Err(jmap_client::Error::Problem(err)) if err.status() == Some(401))); @@ -85,6 +85,7 @@ pub async fn test(params: &mut JMAPTest) { &format!("brute_force{}", n), )) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await { @@ -145,6 +146,7 @@ pub async fn test(params: &mut JMAPTest) { Client::new() .credentials(Credentials::basic("jdoe@example.com", "12345")) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await .unwrap(); @@ -154,6 +156,7 @@ pub async fn test(params: &mut JMAPTest) { let client = Client::new() .credentials(Credentials::basic("jdoe@example.com", "12345")) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await .unwrap(); diff --git a/tests/src/jmap/auth_oauth.rs b/tests/src/jmap/auth_oauth.rs index 538f1c36..fdeb9adf 100644 --- a/tests/src/jmap/auth_oauth.rs +++ b/tests/src/jmap/auth_oauth.rs @@ -145,6 +145,7 @@ pub async fn test(params: &mut JMAPTest) { let john_client = Client::new() .credentials(Credentials::bearer(&token)) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await .unwrap(); @@ -323,6 +324,7 @@ pub async fn test(params: &mut JMAPTest) { let john_client = Client::new() .credentials(Credentials::bearer(&token)) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await .unwrap(); @@ -495,6 +497,7 @@ async fn assert_unauthorized(base_url: &str, token: &str) { match Client::new() .credentials(Credentials::bearer(token)) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect(base_url) .await { diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 899f79cb..4cf2a89b 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -376,6 +376,7 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { .credentials(Credentials::basic("admin", "secret")) .timeout(Duration::from_secs(3600)) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await .unwrap(); @@ -496,6 +497,7 @@ pub async fn test_account_login(login: &str, secret: &str) -> Client { .credentials(Credentials::basic(login, secret)) .timeout(Duration::from_secs(5)) .accept_invalid_certs(true) + .follow_redirects(["127.0.0.1"]) .connect("https://127.0.0.1:8899") .await .unwrap() diff --git a/tests/src/jmap/quota.rs b/tests/src/jmap/quota.rs index 98b30297..1094bcc6 100644 --- a/tests/src/jmap/quota.rs +++ b/tests/src/jmap/quota.rs @@ -10,6 +10,7 @@ use crate::{ assert_is_empty, delivery::SmtpConnection, emails_purge_tombstoned, jmap_raw_request, mailbox::destroy_all_mailboxes, test_account_login, }, + smtp::queue::QueuedEvents, }; use common::config::smtp::queue::QueueName; use email::mailbox::INBOX_ID; @@ -359,7 +360,7 @@ pub async fn test(params: &mut JMAPTest) { params.client.set_default_account_id(account_id.to_string()); destroy_all_mailboxes(params).await; } - for event in server.next_event().await { + for event in server.all_queued_messages().await.messages { server .read_message(event.queue_id, QueueName::default()) .await diff --git a/tests/src/smtp/inbound/mod.rs b/tests/src/smtp/inbound/mod.rs index 31ca925f..5d67206e 100644 --- a/tests/src/smtp/inbound/mod.rs +++ b/tests/src/smtp/inbound/mod.rs @@ -197,6 +197,7 @@ impl QueueReceiver { messages.push(MessageWrapper { queue_id: key.deserialize_be_u64(0)?, queue_name: Default::default(), + is_multi_queue: false, span_id: 0, message: as Deserialize>::deserialize(value)? .deserialize::()?, diff --git a/tests/src/smtp/lookup/utils.rs b/tests/src/smtp/lookup/utils.rs index c20a574f..17f750ce 100644 --- a/tests/src/smtp/lookup/utils.rs +++ b/tests/src/smtp/lookup/utils.rs @@ -11,7 +11,7 @@ use ::smtp::outbound::NextHop; use common::{ Core, config::smtp::{ - queue::MxConfig, + queue::{MxConfig, QueueExpiry, QueueName}, report::AggregateFrequency, resolver::{Mode, MxPattern, Policy}, }, @@ -23,8 +23,13 @@ use smtp::{ lookup::{SourceIp, ToNextHop}, mta_sts::parse::ParsePolicy, }, + queue::{ + Error, ErrorDetails, FROM_AUTHENTICATED, Message, QueueEnvelope, Recipient, Schedule, + Status, + }, reporting::AggregateTimestamp, }; +use store::write::now; use utils::config::Config; const CONFIG: &str = r#" @@ -74,10 +79,13 @@ ip-lookup-strategy = "ipv4_then_ipv6" type = "mx" ip-lookup-strategy = "ipv6_then_ipv4" +[queue.strategy] +schedule = "source + ' ' + received_from_ip + ' ' + received_via_port + ' ' + queue_name + ' ' + last_error + ' ' + rcpt_domain + ' ' + size + ' ' + queue_age" + "#; #[tokio::test] -async fn lookup_ip() { +async fn strategies() { // Enable logging crate::enable_logging(); @@ -137,6 +145,48 @@ async fn lookup_ip() { } } } + + // Test strategy resolution + let message = Message { + created: now() - 123, + blob_hash: Default::default(), + received_from_ip: "1.2.3.4".parse().unwrap(), + received_via_port: 7911, + return_path: "test@example.com".to_string(), + return_path_lcase: "test@example.com".to_string(), + return_path_domain: "example.com".to_string(), + recipients: vec![Recipient { + address: "recipient@foobar.com".to_string(), + address_lcase: "recipient@foobar.com".to_string(), + retry: Schedule::now(), + notify: Schedule::now(), + expires: QueueExpiry::Duration(3600), + queue: QueueName::new("test").unwrap(), + status: Status::TemporaryFailure(ErrorDetails { + entity: "test.example.com".to_string(), + details: Error::TlsError("TLS handshake failed".to_string()), + }), + flags: 0, + orcpt: None, + }], + flags: FROM_AUTHENTICATED, + env_id: None, + priority: 0, + size: 978, + quota_keys: vec![], + }; + + assert_eq!( + test.server + .eval_if::( + &test.server.core.smtp.queue.queue, + &QueueEnvelope::new(&message, &message.recipients[0]), + 0, + ) + .await + .unwrap_or_else(|| "default".to_string()), + "authenticated 1.2.3.4 7911 test tls foobar.com 978 123" + ); } #[test] diff --git a/tests/src/smtp/outbound/lmtp.rs b/tests/src/smtp/outbound/lmtp.rs index ed159a82..39ca21a8 100644 --- a/tests/src/smtp/outbound/lmtp.rs +++ b/tests/src/smtp/outbound/lmtp.rs @@ -9,13 +9,14 @@ use std::time::{Duration, Instant}; use crate::smtp::{ DnsCache, TestSMTP, inbound::TestMessage, + queue::QueuedEvents, session::{TestSession, VerifyResponse}, }; use common::{ config::{server::ServerProtocol, smtp::queue::QueueName}, ipc::QueueEvent, }; -use smtp::queue::spool::SmtpSpool; +use smtp::queue::spool::{QUEUE_REFRESH, SmtpSpool}; use store::write::now; const REMOTE: &str = " @@ -120,20 +121,21 @@ async fn lmtp_delivery() { loop { match local.queue_receiver.try_read_event().await { Some(QueueEvent::Refresh | QueueEvent::WorkerDone { .. }) => {} - Some(QueueEvent::Paused(_)) => unreachable!(), + Some(QueueEvent::Paused(_)) | Some(QueueEvent::ReloadSettings) => unreachable!(), None | Some(QueueEvent::Stop) => break, } - let events = core.next_event().await; - if events.is_empty() { - break; - } - let now = now(); - for event in events { - if event.due > now { - tokio::time::sleep(Duration::from_secs(event.due - now)).await; + let mut events = core.all_queued_messages().await; + if events.messages.is_empty() { + let now = now(); + if events.next_refresh < now + QUEUE_REFRESH { + tokio::time::sleep(Duration::from_secs(events.next_refresh - now)).await; + events = core.all_queued_messages().await; + } else { + break; } - + } + for event in events.messages { let message = core .read_message(event.queue_id, QueueName::default()) .await diff --git a/tests/src/smtp/outbound/smtp.rs b/tests/src/smtp/outbound/smtp.rs index 8fb9ead2..8e30cb15 100644 --- a/tests/src/smtp/outbound/smtp.rs +++ b/tests/src/smtp/outbound/smtp.rs @@ -16,9 +16,10 @@ use store::write::now; use crate::smtp::{ DnsCache, TestSMTP, inbound::{TestMessage, TestQueueEvent}, + queue::QueuedEvents, session::{TestSession, VerifyResponse}, }; -use smtp::queue::spool::SmtpSpool; +use smtp::queue::spool::{QUEUE_REFRESH, SmtpSpool}; const LOCAL: &str = r#" [session.rcpt] @@ -52,6 +53,8 @@ schedule = [{if = "rcpt_domain == 'foobar.org'", then = "'foobar-org'"}, {if = "rcpt_domain == 'foobar.com'", then = "'foobar-com'"}, {else = "'default'"}] +[spam-filter] +enable = false "#; @@ -65,6 +68,10 @@ relay = true [session.extensions] dsn = true chunking = false + +[spam-filter] +enable = false + "#; const SMUGGLER: &str = r#"From: Joe SixPack @@ -158,20 +165,23 @@ async fn smtp_delivery() { loop { match local.queue_receiver.try_read_event().await { Some(QueueEvent::Refresh | QueueEvent::WorkerDone { .. }) => {} - Some(QueueEvent::Paused(_)) => unreachable!(), - None | Some(QueueEvent::Stop) => break, - } - - let events = core.next_event().await; - if events.is_empty() { - break; - } - let now = now(); - for event in events { - if event.due > now { - tokio::time::sleep(Duration::from_secs(event.due - now)).await; + Some(QueueEvent::Paused(_)) | Some(QueueEvent::ReloadSettings) => unreachable!(), + None | Some(QueueEvent::Stop) => { + break; } + } + let mut events = core.all_queued_messages().await; + if events.messages.is_empty() { + let now = now(); + if events.next_refresh < now + QUEUE_REFRESH { + tokio::time::sleep(Duration::from_secs(events.next_refresh - now)).await; + events = core.all_queued_messages().await; + } else { + break; + } + } + for event in events.messages { let message = core .read_message(event.queue_id, QueueName::default()) .await diff --git a/tests/src/smtp/outbound/throttle.rs b/tests/src/smtp/outbound/throttle.rs index 3c1190cd..bf66284b 100644 --- a/tests/src/smtp/outbound/throttle.rs +++ b/tests/src/smtp/outbound/throttle.rs @@ -11,7 +11,7 @@ use crate::smtp::{ session::TestSession, }; use mail_auth::MX; -use smtp::queue::{Message, QueueEnvelope, throttle::IsAllowed}; +use smtp::queue::{DomainPart, Message, QueueEnvelope, Recipient, throttle::IsAllowed}; use std::{ net::{IpAddr, Ipv4Addr}, time::{Duration, Instant}, @@ -69,6 +69,9 @@ async fn throttle_outbound() { // Build test message let mut test_message = new_message(0).message; test_message.return_path_domain = "foobar.org".into(); + test_message + .recipients + .push(build_rcpt("bill@test.org", 0, 0, 0)); let mut local = TestSMTP::new("smtp_throttle_outbound", CONFIG).await; @@ -88,9 +91,13 @@ async fn throttle_outbound() { // Throttle sender let throttle = &core.core.smtp.queue.outbound_limiters; for t in &throttle.sender { - core.is_allowed(t, &QueueEnvelope::test(&test_message, 0, ""), 0) - .await - .unwrap(); + core.is_allowed( + t, + &QueueEnvelope::test(&test_message, &test_message.recipients[0], ""), + 0, + ) + .await + .unwrap(); } // Expect concurrency throttle for sender domain 'foobar.org' @@ -105,10 +112,15 @@ async fn throttle_outbound() { // Expect rate limit throttle for sender domain 'foobar.net' test_message.return_path_domain = "foobar.net".into(); for t in &throttle.sender { - core.is_allowed(t, &QueueEnvelope::test(&test_message, 0, ""), 0) - .await - .unwrap(); + core.is_allowed( + t, + &QueueEnvelope::test(&test_message, &test_message.recipients[0], ""), + 0, + ) + .await + .unwrap(); } + test_message.recipients.clear(); session .send_message("john@foobar.net", &["bill@test.org"], "test:no_dkim", "250") @@ -129,9 +141,13 @@ async fn throttle_outbound() { .recipients .push(build_rcpt("test@example.org", 0, 0, 0)); for t in &throttle.rcpt { - core.is_allowed(t, &QueueEnvelope::test(&test_message, 0, ""), 0) - .await - .unwrap(); + core.is_allowed( + t, + &QueueEnvelope::test(&test_message, &test_message.recipients[0], ""), + 0, + ) + .await + .unwrap(); } /*session @@ -155,9 +171,13 @@ async fn throttle_outbound() { .recipients .push(build_rcpt("test@example.net", 0, 0, 0)); for t in &throttle.rcpt { - core.is_allowed(t, &QueueEnvelope::test(&test_message, 1, ""), 0) - .await - .unwrap(); + core.is_allowed( + t, + &QueueEnvelope::test(&test_message, &test_message.recipients[1], ""), + 0, + ) + .await + .unwrap(); } session @@ -197,9 +217,13 @@ async fn throttle_outbound() { .push(build_rcpt("test@test.org", 0, 0, 0)); for t in &throttle.remote { - core.is_allowed(t, &QueueEnvelope::test(&test_message, 2, "mx.test.org"), 0) - .await - .unwrap(); + core.is_allowed( + t, + &QueueEnvelope::test(&test_message, &test_message.recipients[2], "mx.test.org"), + 0, + ) + .await + .unwrap(); } /*session @@ -227,9 +251,13 @@ async fn throttle_outbound() { Instant::now() + Duration::from_secs(10), ); for t in &throttle.remote { - core.is_allowed(t, &QueueEnvelope::test(&test_message, 1, "mx.test.net"), 0) - .await - .unwrap(); + core.is_allowed( + t, + &QueueEnvelope::test(&test_message, &test_message.recipients[1], "mx.test.net"), + 0, + ) + .await + .unwrap(); } session @@ -248,17 +276,18 @@ async fn throttle_outbound() { } pub trait TestQueueEnvelope<'x> { - fn test(message: &'x Message, current_domain: usize, mx: &'x str) -> Self; + fn test(message: &'x Message, rcpt: &'x Recipient, mx: &'x str) -> Self; } impl<'x> TestQueueEnvelope<'x> for QueueEnvelope<'x> { - fn test(message: &'x Message, current_rcpt: usize, mx: &'x str) -> Self { + fn test(message: &'x Message, rcpt: &'x Recipient, mx: &'x str) -> Self { QueueEnvelope { message, mx, remote_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), local_ip: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), - current_rcpt, + domain: rcpt.address_lcase.domain_part(), + rcpt, } } } diff --git a/tests/src/smtp/queue/dsn.rs b/tests/src/smtp/queue/dsn.rs index 8655dc16..75c4d5d9 100644 --- a/tests/src/smtp/queue/dsn.rs +++ b/tests/src/smtp/queue/dsn.rs @@ -56,6 +56,7 @@ async fn generate_dsn() { let mut message = MessageWrapper { queue_id: 0, span_id: 0, + is_multi_queue: false, queue_name: QueueName::default(), message: Message { size, diff --git a/tests/src/smtp/queue/manager.rs b/tests/src/smtp/queue/manager.rs index 1e0a982f..1fe3ff0f 100644 --- a/tests/src/smtp/queue/manager.rs +++ b/tests/src/smtp/queue/manager.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::smtp::{TestSMTP, queue::build_rcpt}; +use crate::smtp::{ + TestSMTP, + queue::{QueuedEvents, build_rcpt}, +}; use common::config::smtp::queue::QueueName; use smtp::queue::{ Error, ErrorDetails, Message, MessageWrapper, Recipient, Status, spool::SmtpSpool, @@ -46,12 +49,14 @@ async fn queue_due() { for domain in vec!["a", "b", "c"].into_iter() { let now = now(); - for queue_event in core.next_event().await { - if queue_event.due > now { - let wake_up = queue_event.due - now; - assert_eq!(wake_up, 1); - std::thread::sleep(Duration::from_secs(wake_up)); - } + let queued = core.all_queued_messages().await; + if queued.messages.is_empty() { + let wake_up = queued.next_refresh - now; + assert_eq!(wake_up, 1); + std::thread::sleep(Duration::from_secs(wake_up)); + } + + for queue_event in core.all_queued_messages().await.messages { if let Some(message) = core .read_message(queue_event.queue_id, QueueName::default()) .await @@ -86,36 +91,34 @@ fn delivery_events() { message.rcpt("a").retry.due ); assert_eq!( - message - .next_event_after( - None, - message.rcpt("a").expiration_time(message.created).unwrap() - ) - .unwrap(), + next_event_after( + &message, + None, + message.rcpt("a").expiration_time(message.created).unwrap() + ) + .unwrap(), message.rcpt("b").retry.due ); assert_eq!( - message - .next_event_after( - None, - message.rcpt("b").expiration_time(message.created).unwrap() - ) - .unwrap(), + next_event_after( + &message, + None, + message.rcpt("b").expiration_time(message.created).unwrap() + ) + .unwrap(), message.rcpt("c").retry.due ); assert_eq!( - message - .next_event_after(None, message.rcpt("c").notify.due) - .unwrap(), + next_event_after(&message, None, message.rcpt("c").notify.due).unwrap(), message.rcpt("c").expiration_time(message.created).unwrap() ); assert!( - message - .next_event_after( - None, - message.rcpt("c").expiration_time(message.created).unwrap() - ) - .is_none() + next_event_after( + &message, + None, + message.rcpt("c").expiration_time(message.created).unwrap() + ) + .is_none() ); if t == 0 { @@ -163,6 +166,7 @@ pub fn new_message(queue_id: u64) -> MessageWrapper { queue_id, span_id: 0, queue_name: QueueName::default(), + is_multi_queue: false, message: Message { size: 0, created: now(), @@ -181,6 +185,34 @@ pub fn new_message(queue_id: u64) -> MessageWrapper { } } +fn next_event_after(message: &Message, queue: Option, instant: u64) -> Option { + let mut next_event = None; + + for rcpt in &message.recipients { + if matches!(rcpt.status, Status::Scheduled | Status::TemporaryFailure(_)) + && queue.is_none_or(|q| rcpt.queue == q) + { + if rcpt.retry.due > instant + && next_event.as_ref().is_none_or(|ne| rcpt.retry.due.lt(ne)) + { + next_event = rcpt.retry.due.into(); + } + if rcpt.notify.due > instant + && next_event.as_ref().is_none_or(|ne| rcpt.notify.due.lt(ne)) + { + next_event = rcpt.notify.due.into(); + } + if let Some(expires) = rcpt.expiration_time(message.created) { + if expires > instant && next_event.as_ref().is_none_or(|ne| expires.lt(ne)) { + next_event = expires.into(); + } + } + } + } + + next_event +} + pub trait TestMessage { fn rcpt(&self, name: &str) -> &Recipient; fn rcpt_mut(&mut self, name: &str) -> &mut Recipient; diff --git a/tests/src/smtp/queue/mod.rs b/tests/src/smtp/queue/mod.rs index 4276244d..49e86952 100644 --- a/tests/src/smtp/queue/mod.rs +++ b/tests/src/smtp/queue/mod.rs @@ -4,13 +4,22 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use common::config::smtp::queue::{QueueExpiry, QueueName}; -use smtp::queue::{Recipient, Schedule, Status}; +use common::{ + Server, + config::smtp::queue::{QueueExpiry, QueueName}, +}; +use smtp::queue::{ + Recipient, Schedule, Status, + manager::Queue, + spool::{QueuedMessages, SmtpSpool}, +}; +use tokio::sync::mpsc; pub mod concurrent; pub mod dsn; pub mod manager; pub mod retry; +pub mod virtualq; pub fn build_rcpt(address: &str, retry: u64, notify: u64, expires: u64) -> Recipient { Recipient { @@ -25,3 +34,14 @@ pub fn build_rcpt(address: &str, retry: u64, notify: u64, expires: u64) -> Recip queue: QueueName::default(), } } + +pub trait QueuedEvents: Sync + Send { + fn all_queued_messages(&self) -> impl Future + Send; +} + +impl QueuedEvents for Server { + async fn all_queued_messages(&self) -> QueuedMessages { + self.next_event(&mut Queue::new(self.inner.clone(), mpsc::channel(100).1)) + .await + } +} diff --git a/tests/src/smtp/queue/retry.rs b/tests/src/smtp/queue/retry.rs index 8c218e88..e17847b4 100644 --- a/tests/src/smtp/queue/retry.rs +++ b/tests/src/smtp/queue/retry.rs @@ -9,6 +9,7 @@ use std::time::Duration; use crate::smtp::{ TestSMTP, inbound::{TestMessage, TestQueueEvent}, + queue::QueuedEvents, session::{TestSession, VerifyResponse}, }; use ahash::AHashSet; @@ -16,7 +17,7 @@ use common::{ config::smtp::queue::QueueName, ipc::{QueueEvent, QueueEventStatus}, }; -use smtp::queue::spool::SmtpSpool; +use smtp::queue::spool::{QUEUE_REFRESH, SmtpSpool}; use store::write::now; const CONFIG: &str = r#" @@ -105,31 +106,34 @@ async fn queue_retry() { loop { match qr.try_read_event().await { - Some(QueueEvent::WorkerDone { queue_id, status }) => { + Some(QueueEvent::WorkerDone { + queue_id, status, .. + }) => { in_fight.remove(&queue_id); match &status { QueueEventStatus::Completed | QueueEventStatus::Deferred => (), _ => panic!("unexpected status {queue_id}: {status:?}"), } } - Some(QueueEvent::Refresh) => (), + Some(QueueEvent::Refresh) | Some(QueueEvent::ReloadSettings) => (), None | Some(QueueEvent::Stop) | Some(QueueEvent::Paused(_)) => break, } let now = now(); - let events = core.next_event().await; - - if events.is_empty() && in_fight.is_empty() { - break; + let mut events = core.all_queued_messages().await; + if events.messages.is_empty() { + if events.next_refresh < now + QUEUE_REFRESH { + tokio::time::sleep(Duration::from_secs(events.next_refresh - now)).await; + events = core.all_queued_messages().await; + } else if in_fight.is_empty() { + break; + } } - for event in events { + for event in events.messages { if in_fight.contains(&event.queue_id) { continue; } - if event.due > now { - tokio::time::sleep(Duration::from_secs(event.due - now)).await; - } let message = core .read_message(event.queue_id, QueueName::default()) diff --git a/tests/src/smtp/queue/virtualq.rs b/tests/src/smtp/queue/virtualq.rs new file mode 100644 index 00000000..1847257d --- /dev/null +++ b/tests/src/smtp/queue/virtualq.rs @@ -0,0 +1,213 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::time::{Duration, Instant}; + +use common::{ + config::{server::ServerProtocol, smtp::queue::QueueName}, + core::BuildServer, + ipc::QueueEvent, +}; +use mail_auth::MX; + +use crate::smtp::{DnsCache, TestSMTP, session::TestSession}; +use smtp::queue::manager::Queue; + +const LOCAL: &str = r#" +[spam-filter] +enable = false + +[session.rcpt] +relay = true + +[session.data.limits] +messages = 2000 + +[queue.strategy] +schedule = [ { if = "rcpt == 'delay-random@foobar.org'", then = "'q2'" }, + { else = "'q1'"} ] + +[queue.virtual.q1] +threads-per-node = 5 + +[queue.virtual.q2] +threads-per-node = 4 + +[queue.schedule.q1] +retry = "1s" +notify = "1d" +expire = "1d" +queue-name = "q1" + +[queue.schedule.q2] +retry = "1s" +notify = "1d" +expire = "1d" +queue-name = "q2" + +"#; + +const REMOTE: &str = r#" +[session.ehlo] +reject-non-fqdn = false + +[session.rcpt] +relay = true + +[spam-filter] +enable = false + +"#; + +const NUM_MESSAGES: usize = 100; +const NUM_QUEUES: usize = 10; + +#[tokio::test(flavor = "multi_thread", worker_threads = 18)] +#[serial_test::serial] +async fn virtual_queue() { + // Enable logging + crate::enable_logging(); + + // Start test server + let remote = TestSMTP::new("smtp_virtual_queue_remote", REMOTE).await; + let _rx = remote.start(&[ServerProtocol::Smtp]).await; + let local = TestSMTP::with_database("smtp_virtual_queue_local", LOCAL, "mysql").await; + + // Validate parsing + for value in ["a", "ab", "abcdefgh"] { + let queue_name = QueueName::new(value).unwrap(); + assert_eq!(queue_name.to_string(), value); + } + assert_eq!( + local + .server + .core + .smtp + .queue + .virtual_queues + .get(&QueueName::new("q1").unwrap()) + .unwrap() + .threads, + 5 + ); + assert_eq!( + local + .server + .core + .smtp + .queue + .virtual_queues + .get(&QueueName::new("q2").unwrap()) + .unwrap() + .threads, + 4 + ); + + // Add mock DNS entries + let core = local.build_smtp(); + core.mx_add( + "foobar.org", + vec![MX { + exchanges: vec!["mx.foobar.org".to_string()], + preference: 10, + }], + Instant::now() + Duration::from_secs(100), + ); + core.ipv4_add( + "mx.foobar.org", + vec!["127.0.0.1".parse().unwrap()], + Instant::now() + Duration::from_secs(100), + ); + + let mut session = local.new_session(); + session.data.remote_ip_str = "10.0.0.1".into(); + session.eval_session_params().await; + session.ehlo("mx.test.org").await; + + // Spawn concurrent queues + let mut inners = vec![]; + for _ in 0..NUM_QUEUES { + let (inner, rxs) = local.inner_with_rxs(); + let server = inner.build_server(); + server.mx_add( + "foobar.org", + vec![MX { + exchanges: vec!["mx.foobar.org".to_string()], + preference: 10, + }], + Instant::now() + Duration::from_secs(100), + ); + server.ipv4_add( + "mx.foobar.org", + vec!["127.0.0.1".parse().unwrap()], + Instant::now() + Duration::from_secs(100), + ); + inners.push(inner.clone()); + tokio::spawn(async move { + Queue::new(inner, rxs.queue_rx.unwrap()).start().await; + }); + } + + tokio::time::sleep(Duration::from_millis(200)).await; + + // Send 1000 test messages + for _ in 0..(NUM_MESSAGES / 2) { + session + .send_message( + "john@test.org", + &["bill@foobar.org", "delay-random@foobar.org"], + "test:no_dkim", + "250", + ) + .await; + } + + // Wake up all queues + for inner in &inners { + inner.ipc.queue_tx.send(QueueEvent::Refresh).await.unwrap(); + } + for _ in 0..(NUM_MESSAGES / 2) { + session + .send_message( + "john@test.org", + &["bill@foobar.org", "delay-random@foobar.org"], + "test:no_dkim", + "250", + ) + .await; + } + + loop { + tokio::time::sleep(Duration::from_millis(1500)).await; + + let m = local.queue_receiver.read_queued_messages().await; + let e = local.queue_receiver.read_queued_events().await; + + if m.len() + e.len() != 0 { + println!( + "Queue still has {} messages and {} events", + m.len(), + e.len() + ); + /*for inner in &inners { + inner.ipc.queue_tx.send(QueueEvent::Refresh).await.unwrap(); + }*/ + } else { + break; + } + } + + local.queue_receiver.assert_queue_is_empty().await; + let remote_messages = remote.queue_receiver.read_queued_messages().await; + assert_eq!(remote_messages.len(), NUM_MESSAGES * 2); + + // Make sure local store is queue + core.core + .storage + .data + .assert_is_empty(core.core.storage.blob.clone()) + .await; +} diff --git a/tests/src/store/ops.rs b/tests/src/store/ops.rs index bd215ec4..eb7c5f96 100644 --- a/tests/src/store/ops.rs +++ b/tests/src/store/ops.rs @@ -98,6 +98,49 @@ pub async fn test(db: Store) { db.write(batch.build_all()).await.unwrap(); } + // Merge values 1000 times concurrently + let mut handles = Vec::new(); + println!("Merge values 1000 times concurrently..."); + for _ in 0..1000 { + handles.push({ + let db = db.clone(); + tokio::spawn(async move { + let mut builder = BatchBuilder::new(); + builder + .with_account_id(0) + .with_collection(0) + .update_document(0) + .merge(ValueClass::Property(3), |bytes| { + if let Some(bytes) = bytes { + Ok((u64::from_be_bytes(bytes.try_into().unwrap()) + 1) + .to_be_bytes() + .to_vec()) + } else { + Ok(0u64.to_be_bytes().to_vec()) + } + }); + db.write(builder.build_all()).await.unwrap() + }) + }); + } + + for handle in handles { + handle.await.unwrap(); + } + + assert_eq!( + 999, + db.get_value::(ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Property(3), + }) + .await + .unwrap() + .unwrap() + ); + // Increment a counter 1000 times concurrently let mut handles = Vec::new(); let mut assigned_ids = HashSet::new(); @@ -305,6 +348,7 @@ pub async fn test(db: Store) { .update_document(0) .clear(ValueClass::Property(0)) .clear(ValueClass::Property(2)) + .clear(ValueClass::Property(3)) .clear(ValueClass::Directory(DirectoryClass::UsedQuota(0))) .clear(ValueClass::ChangeId);