diff --git a/crates/common/src/cache/reload.rs b/crates/common/src/cache/reload.rs index f75e968..6efa4f0 100644 --- a/crates/common/src/cache/reload.rs +++ b/crates/common/src/cache/reload.rs @@ -20,13 +20,20 @@ use ahash::AHashMap; use directory::Directories; use registry::{ schema::{prelude::ObjectType, structs::BlockedIp}, - types::error::{Error, Warning}, + types::{ + error::{Error, Warning}, + id::ObjectId, + }, }; use std::sync::Arc; use store::{LookupStores, registry::bootstrap::Bootstrap, write::now}; pub struct ReloadResult { + /// Errors that kept the reload from being applied. pub errors: Vec, + /// inbuxa: errors in objects that already failed when the running + /// settings were built; logged, but they don't refuse a reload. + pub known_errors: Vec, pub warnings: Vec, pub replaced_core: bool, } @@ -114,42 +121,60 @@ impl Server { directories: directory.directories, }; - // Parse tracers + // inbuxa: upstream swapped the core only when the whole build + // was free of errors, while boot runs with whatever built. So one + // object that failed (a DNS lookup that timed out, say) refused + // every later reload, cluster-wide when the reload came from + // ReloadSettings, and the running settings went stale. Now a + // reload is refused only for errors in objects that built when + // the running settings were built: those would be lost by + // applying it. Objects that already failed then are missing + // from the running settings anyway, as at boot, so their + // errors are reported but don't hold the reload back. let tracers = Telemetry::parse(&mut bootstrap, &storage).await; + let core = Box::pin(Core::parse(&mut bootstrap, storage)).await; + let mut servers = Listeners::parse(&mut bootstrap).await; - if bootstrap.errors.is_empty() { - let core = Box::pin(Core::parse(&mut bootstrap, storage)).await; + if !self.has_new_build_errors(&bootstrap.errors) { + servers + .parse_tcp_acceptors(&mut bootstrap, self.inner.clone()) + .await; - if bootstrap.errors.is_empty() { - let mut servers = Listeners::parse(&mut bootstrap).await; - servers - .parse_tcp_acceptors(&mut bootstrap, self.inner.clone()) - .await; + if !self.has_new_build_errors(&bootstrap.errors) { + // Update core + self.inner.shared_core.store(core.into()); - if bootstrap.errors.is_empty() { - // Update core - self.inner.shared_core.store(core.into()); + // Update tracers + tracers.update(); - // Update tracers + // Reload queue settings + self.inner + .ipc + .queue_tx + .send(QueueEvent::ReloadSettings) + .await + .ok(); - tracers.update(); + self.record_build_errors(&bootstrap.errors); - // Reload queue settings - self.inner - .ipc - .queue_tx - .send(QueueEvent::ReloadSettings) - .await - .ok(); - - return Ok(ReloadResult { - errors: bootstrap.errors, - warnings: bootstrap.warnings, - replaced_core: true, - }); - } + return Ok(ReloadResult { + errors: Vec::new(), + known_errors: bootstrap.errors, + warnings: bootstrap.warnings, + replaced_core: true, + }); } } + + let (known_errors, errors) = std::mem::take(&mut bootstrap.errors) + .into_iter() + .partition(|error| self.is_known_build_error(error)); + return Ok(ReloadResult { + errors, + known_errors, + warnings: bootstrap.warnings, + replaced_core: false, + }); } } @@ -163,7 +188,7 @@ impl ReloadResult { } pub fn log(&self) { - for error in &self.errors { + for error in self.errors.iter().chain(&self.known_errors) { error.log(); } for warning in &self.warnings { @@ -176,8 +201,34 @@ impl From for ReloadResult { fn from(bootstrap: Bootstrap) -> Self { Self { errors: bootstrap.errors, + known_errors: Vec::new(), warnings: bootstrap.warnings, replaced_core: false, } } } + +// inbuxa: which objects failed to build for the running settings +impl Server { + /// Records the objects that failed to build for the settings now running. + pub fn record_build_errors(&self, errors: &[Error]) { + *self.inner.data.build_errors.lock() = errors.iter().filter_map(error_object).collect(); + } + + fn is_known_build_error(&self, error: &Error) -> bool { + error_object(error).is_some_and(|id| self.inner.data.build_errors.lock().contains(&id)) + } + + fn has_new_build_errors(&self, errors: &[Error]) -> bool { + errors.iter().any(|error| !self.is_known_build_error(error)) + } +} + +fn error_object(error: &Error) -> Option { + match error { + Error::Validation { object_id, .. } + | Error::Build { object_id, .. } + | Error::NotFound { object_id } => Some(*object_id), + Error::Internal { object_id, .. } => *object_id, + } +} diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index b851e26..094e73e 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -96,6 +96,7 @@ impl Data { applications, logos: Default::default(), smtp_connectors: TlsConnectors::try_new().failed("Failed to build TLS connectors"), + build_errors: Default::default(), asn_geo_data: Default::default(), } } @@ -237,6 +238,7 @@ impl Default for Data { applications: WebApplications::new(), logos: Default::default(), smtp_connectors: TlsConnectors::try_new().unwrap(), + build_errors: Default::default(), asn_geo_data: Default::default(), lookup_stores: Default::default(), } diff --git a/crates/common/src/config/mailstore/spamfilter.rs b/crates/common/src/config/mailstore/spamfilter.rs index 779e234..811efbb 100644 --- a/crates/common/src/config/mailstore/spamfilter.rs +++ b/crates/common/src/config/mailstore/spamfilter.rs @@ -16,7 +16,6 @@ use mail_auth::common::resolver::ToReverseName; use nlp::classifier::model::{CcfhClassifier, FhClassifier}; use registry::schema::{ enums::{ExpressionVariable, ModelSize}, - prelude::ObjectType, structs::{ self, SpamDnsblServer, SpamDnsblSettings, SpamFileExtension, SpamPyzor, SpamRule, SpamSettings, SpamTag, @@ -25,10 +24,10 @@ use registry::schema::{ use sieve::SpamStatus; use std::{ net::{IpAddr, SocketAddr}, - time::Duration, + sync::Arc, + time::{Duration, Instant}, }; use store::registry::{RegistryObject, bootstrap::Bootstrap}; -use tokio::net::lookup_host; use utils::{cache::CacheItemWeight, glob::GlobMap}; #[derive(rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default)] @@ -157,7 +156,11 @@ pub struct FtrlParameters { #[derive(Debug, Clone)] pub struct PyzorConfig { - pub address: SocketAddr, + // inbuxa: the server is resolved when a message is checked, not while the + // settings are built (see PyzorConfig::address) + pub host: String, + pub port: u16, + pub resolved: Arc>>, pub timeout: Duration, pub min_count: u64, pub min_wl_count: u64, @@ -474,31 +477,15 @@ impl PyzorConfig { return None; } - let port = pyzor.port; - let host = pyzor.host; - let address = match lookup_host(format!("{host}:{port}")) - .await - .map(|mut a| a.next()) - { - Ok(Some(address)) => address, - Ok(None) => { - bp.build_error( - ObjectType::SpamPyzor.singleton(), - "Invalid address: No addresses found.", - ); - return None; - } - Err(err) => { - bp.build_error( - ObjectType::SpamPyzor.singleton(), - format!("Invalid address: {}", err), - ); - return None; - } - }; - + // inbuxa: upstream resolved the host here and reported a failed lookup + // as a build error, so a DNS hiccup on one node refused every settings + // reload on it (and, from the node that ran ReloadSettings, across the + // cluster). The lookup now happens when a message is checked; a + // failure there is logged as a Pyzor error for that message. PyzorConfig { - address, + host: pyzor.host, + port: pyzor.port as u16, + resolved: Default::default(), timeout: pyzor.timeout.into_inner(), min_count: pyzor.block_count, min_wl_count: pyzor.allow_count, @@ -508,6 +495,35 @@ impl PyzorConfig { } } +// inbuxa: how long a resolved Pyzor address is reused +const PYZOR_RESOLVE_TTL: Duration = Duration::from_secs(300); + +impl PyzorConfig { + /// The server's address: the host itself when it is an IP address, + /// otherwise the first address it resolves to, reused for five minutes. + pub async fn address(&self) -> std::io::Result { + if let Ok(ip) = self.host.parse::() { + return Ok(SocketAddr::new(ip, self.port)); + } + if let Some((address, resolved_at)) = *self.resolved.lock() + && resolved_at.elapsed() < PYZOR_RESOLVE_TTL + { + return Ok(address); + } + let address = tokio::net::lookup_host((self.host.as_str(), self.port)) + .await? + .next() + .ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::NotFound, + format!("{} has no addresses", self.host), + ) + })?; + *self.resolved.lock() = Some((address, Instant::now())); + Ok(address) + } +} + impl ClassifierConfig { pub async fn parse(bp: &mut Bootstrap) -> Option { let classifier = bp.setting_infallible::().await; diff --git a/crates/common/src/config/smtp/session.rs b/crates/common/src/config/smtp/session.rs index 2efcb13..6890e25 100644 --- a/crates/common/src/config/smtp/session.rs +++ b/crates/common/src/config/smtp/session.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use self::resolver::Policy; @@ -22,7 +24,7 @@ use registry::schema::{ }; use smtp_proto::*; use std::{ - net::{SocketAddr, ToSocketAddrs}, + net::{IpAddr, SocketAddr}, str::FromStr, time::Duration, }; @@ -384,19 +386,16 @@ impl SessionConfig { Some(Milter { enable: bp.compile_expr(id, &milter.ctx_enable()), id, - addrs: format!("{}:{}", milter.hostname, milter.port) - .to_socket_addrs() - .map_err(|err| { - bp.build_error( - id, - format!( - "Unable to resolve milter hostname {}: {}", - milter.hostname, err - ), - ) - }) - .ok()? - .collect(), + // inbuxa: upstream resolved the hostname here (a + // blocking lookup) and made a failure a build error, + // which refused the whole settings reload. An IP + // address is kept as is; a name is resolved on each + // connection (MilterClient::connect). + addrs: milter + .hostname + .parse::() + .map(|ip| vec![SocketAddr::new(ip, milter.port as u16)]) + .unwrap_or_default(), hostname: milter.hostname, port: milter.port as u16, timeout_connect: milter.timeout_connect.into_inner(), diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index d797b14..ad0180a 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -166,6 +166,10 @@ pub struct Data { pub logos: Mutex, LogoCache>>, pub smtp_connectors: TlsConnectors, + + // inbuxa: the objects that failed to build when the running settings + // were built, at boot or by the last applied reload (see reload_registry) + pub build_errors: Mutex>, } #[derive(Clone)] diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index b4e7b13..877e163 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -240,6 +240,9 @@ impl BootManager { .parse_tcp_acceptors(&mut bootstrap, inner.clone()) .await; + // inbuxa: a reload isn't refused over objects that failed here + inner.build_server().record_build_errors(&bootstrap.errors); + BootManager { inner, bootstrap, diff --git a/crates/jmap/src/registry/mapping/action.rs b/crates/jmap/src/registry/mapping/action.rs index 88ece99..800a889 100644 --- a/crates/jmap/src/registry/mapping/action.rs +++ b/crates/jmap/src/registry/mapping/action.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::registry::mapping::{RegistrySetResponse, map_bootstrap_error}; @@ -99,7 +101,7 @@ pub(crate) async fn action_set( } else { set.response .not_created - .append(id, map_bootstrap_error(result.errors)); + .append(id, reload_refused(result.errors)); } } Action::InvalidateCaches => { @@ -573,3 +575,34 @@ async fn dmarc_troubleshoot( Some(request) } + +/// inbuxa: a refused reload names the object that stopped it and says the +/// settings weren't applied; upstream passed on the first error's bare message +/// ("Invalid address: ..."), which read like a problem with the request. +fn reload_refused(errors: Vec) -> SetError { + use registry::types::error::Error; + let more = errors.len().saturating_sub(1); + let mut description = match errors.first() { + Some(Error::Build { object_id, message }) => format!("{object_id}: {message}"), + Some(Error::Validation { object_id, errors }) => format!( + "{object_id}: {}", + errors + .iter() + .map(|err| err.to_string()) + .collect::>() + .join("; ") + ), + Some(Error::Internal { + object_id: Some(object_id), + error, + }) => format!("{object_id}: {error}"), + Some(Error::Internal { error, .. }) => error.to_string(), + Some(Error::NotFound { object_id }) => format!("{object_id} was not found"), + None => String::new(), + }; + description.insert_str(0, "Settings were not reloaded. "); + if more > 0 { + description.push_str(&format!(" ({more} more in the server log.)")); + } + map_bootstrap_error(errors).with_description(description) +} diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index d7a1634..1244219 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -947,10 +947,14 @@ impl RegistrySet for Server { self.cluster_broadcast(common::ipc::BroadcastEvent::RegistryChange(change)) .await; } - Ok(_) => trc::event!( - Registry(trc::RegistryEvent::BuildWarning), - Details = "Settings didn't reload after a directory change", - ), + Ok(reload) => { + // inbuxa: name what stopped it + reload.log(); + trc::event!( + Registry(trc::RegistryEvent::BuildWarning), + Details = "Settings didn't reload after a directory change", + ) + } Err(err) => { trc::error!(err.details("Failed to reload directories")); } diff --git a/crates/smtp/src/inbound/milter/client.rs b/crates/smtp/src/inbound/milter/client.rs index fb86cda..df78929 100644 --- a/crates/smtp/src/inbound/milter/client.rs +++ b/crates/smtp/src/inbound/milter/client.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use common::config::smtp::session::Milter; @@ -25,7 +27,19 @@ impl MilterClient { pub async fn connect(config: &Milter, session_id: u64) -> Result { tokio::time::timeout(config.timeout_command, async { let mut last_err = Error::Disconnected; - for addr in &config.addrs { + // inbuxa: a hostname is resolved here, per connection, rather + // than while the settings are built + let resolved; + let addrs = if config.addrs.is_empty() { + resolved = tokio::net::lookup_host((config.hostname.as_str(), config.port)) + .await + .map_err(Error::Io)? + .collect::>(); + &resolved + } else { + &config.addrs + }; + for addr in addrs { match TcpStream::connect(addr).await { Ok(stream) => { return Ok(MilterClient { diff --git a/crates/spam-filter/src/modules/pyzor.rs b/crates/spam-filter/src/modules/pyzor.rs index bffdc6b..6ee2056 100644 --- a/crates/spam-filter/src/modules/pyzor.rs +++ b/crates/spam-filter/src/modules/pyzor.rs @@ -48,19 +48,24 @@ pub(crate) async fn pyzor_check( // Send message to address. inbuxa: in tests, a fixed table answers // instead of a public server (test_response). #[cfg(not(feature = "test_mode"))] - let response = pyzor_send_message(config.address, config.timeout, &request).await; + let response = match tokio::time::timeout(config.timeout, config.address()).await { + Ok(Ok(address)) => pyzor_send_message(address, config.timeout, &request).await, + Ok(Err(err)) => Err(err), + Err(_) => Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "Timed out resolving the Pyzor server", + )), + }; #[cfg(feature = "test_mode")] let response = std::io::Result::Ok(test_response(&request)); - response - .map(Into::into) - .map_err(|err| { - trc::SpamEvent::PyzorError - .into_err() - .ctx(trc::Key::Url, config.address.to_string()) - .reason(err) - .details("Pyzor failed") - }) + response.map(Into::into).map_err(|err| { + trc::SpamEvent::PyzorError + .into_err() + .ctx(trc::Key::Url, format!("{}:{}", config.host, config.port)) + .reason(err) + .details("Pyzor failed") + }) } /// inbuxa: the answers tests get, by digest, instead of a public server's, diff --git a/tests/src/smtp/inbound/milter.rs b/tests/src/smtp/inbound/milter.rs index 3fc4443..dc64141 100644 --- a/tests/src/smtp/inbound/milter.rs +++ b/tests/src/smtp/inbound/milter.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{ @@ -75,7 +77,7 @@ async fn milter_session() { else_: "true".into(), ..Default::default() }, - hostname: "127.0.0.1".into(), + hostname: "localhost".into(), // inbuxa: resolved when the session connects port: 9332, use_tls: false, stages: Map::new(vec![MtaStage::Data]), diff --git a/tests/src/system/mod.rs b/tests/src/system/mod.rs index e328bf5..8912e07 100644 --- a/tests/src/system/mod.rs +++ b/tests/src/system/mod.rs @@ -20,6 +20,7 @@ pub mod monitoring; pub mod oidc; pub mod purge; pub mod quota; +pub mod reload; // inbuxa: reloads and build errors pub mod security; pub mod task; pub mod tenant; diff --git a/tests/src/system/reload.rs b/tests/src/system/reload.rs new file mode 100644 index 0000000..56c01d7 --- /dev/null +++ b/tests/src/system/reload.rs @@ -0,0 +1,213 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +// inbuxa: a settings reload isn't held back by a DNS lookup, or by objects +// that already failed when the running settings were built; an error in an +// object that built then still refuses it, and says which object. + +use crate::utils::server::{TestServer, TestServerBuilder}; +use common::{BuildServer, config::mailstore::spamfilter::PyzorConfig, ipc::RegistryChange}; +use registry::schema::{ + enums::TracingLevel, + prelude::{ObjectType, Property}, + structs::{Action, Expression, MtaStageAuth, SpamPyzor, Tracer, TracerStdout}, +}; + +#[tokio::test(flavor = "multi_thread")] +pub async fn reload_tests() { + let mut test = TestServerBuilder::new("reload_tests") + .await + .with_default_listeners() + .await + .with_object(MtaStageAuth { + require: Expression { + else_: "false".to_string(), + ..Default::default() + }, + ..Default::default() + }) + .await + .build() + .await; + + let admin = test + .create_user_account( + "admin", + "admin@example.org", + "these_pretzels_are_making_me_thirsty", + &[], + "Admin", + ) + .await; + test.account("admin") + .assign_roles_to_account(admin.id(), &["user", "system"]) + .await; + test.insert_account(admin); + + test_unresolvable_pyzor(&test).await; + test_build_errors(&test).await; + + if test.is_reset() { + test.temp_dir.delete(); + } +} + +async fn test_unresolvable_pyzor(test: &TestServer) { + println!("Running reload with an unresolvable Pyzor host..."); + let admin = test.account("admin@example.org"); + + // Upstream resolved the host while building the settings and refused the + // reload when that failed. + admin + .registry_update_setting( + SpamPyzor { + enable: true, + host: "pyzor.invalid".into(), + port: 24441, + ..Default::default() + }, + &[Property::Enable, Property::Host, Property::Port], + ) + .await; + admin.reload_settings().await; + + let pyzor = running_pyzor(test); + assert_eq!(pyzor.host, "pyzor.invalid"); + assert_eq!(pyzor.port, 24441); + assert!(pyzor.address().await.is_err()); + + // An IP address needs no lookup + admin + .registry_update_setting( + SpamPyzor { + host: "192.0.2.1".into(), + ..Default::default() + }, + &[Property::Host], + ) + .await; + admin.reload_settings().await; + assert_eq!( + running_pyzor(test).address().await.unwrap().to_string(), + "192.0.2.1:24441" + ); +} + +async fn test_build_errors(test: &TestServer) { + println!("Running reload with build errors..."); + let admin = test.account("admin@example.org"); + let pyzor_ratio = running_pyzor(test).ratio; + assert_ne!(pyzor_ratio, 0.25); + + // Two console tracers: only one is allowed, so the build of one of them + // fails. Neither existed when the running settings were built. + let mut tracer_ids = Vec::new(); + for _ in 0..2 { + tracer_ids.push( + admin + .registry_create_object(Tracer::Stdout(TracerStdout { + enable: true, + level: TracingLevel::Error, + ..Default::default() + })) + .await, + ); + } + admin + .registry_update_setting( + SpamPyzor { + ratio: 0.25.into(), + ..Default::default() + }, + &[Property::Ratio], + ) + .await; + + // A new error refuses the reload and names the object + let err = admin + .registry_create_object_expect_err(Action::ReloadSettings) + .await; + let description = err.description.clone().unwrap_or_default(); + assert!( + description.starts_with("Settings were not reloaded. ") + && description.contains("Tracer") + && description.contains("Only one console tracer is allowed"), + "{err:?}" + ); + assert_eq!(running_pyzor(test).ratio, pyzor_ratio); + + // Had the running settings been built with that tracer failing, as a + // restart now would, the same error doesn't hold the reload back. + let result = Box::pin( + test.server + .reload_registry(RegistryChange::Reload(ObjectType::DataStore)), + ) + .await + .unwrap(); + assert!(!result.replaced_core); + assert_eq!(result.errors.len(), 1, "{:?}", result.errors); + test.server.record_build_errors(&result.errors); + + admin.reload_settings().await; + assert_eq!(running_pyzor(test).ratio, 0.25); + let result = Box::pin( + test.server + .reload_registry(RegistryChange::Reload(ObjectType::DataStore)), + ) + .await + .unwrap(); + assert!(result.replaced_core); + assert!(result.errors.is_empty()); + assert_eq!(result.known_errors.len(), 1); + + // Once fixed, the object is no longer known to fail, so a new error + // there refuses the reload again. + admin + .registry_destroy(ObjectType::Tracer, tracer_ids.iter()) + .await + .assert_destroyed(&tracer_ids); + admin.reload_settings().await; + let result = Box::pin( + test.server + .reload_registry(RegistryChange::Reload(ObjectType::DataStore)), + ) + .await + .unwrap(); + assert!(result.replaced_core); + assert!(result.errors.is_empty() && result.known_errors.is_empty()); + + for _ in 0..2 { + tracer_ids.push( + admin + .registry_create_object(Tracer::Stdout(TracerStdout { + enable: true, + level: TracingLevel::Error, + ..Default::default() + })) + .await, + ); + } + admin + .registry_create_object_expect_err(Action::ReloadSettings) + .await; + let tracer_ids = tracer_ids.split_off(2); + admin + .registry_destroy(ObjectType::Tracer, tracer_ids.iter()) + .await + .assert_destroyed(&tracer_ids); + admin.reload_settings().await; +} + +fn running_pyzor(test: &TestServer) -> PyzorConfig { + test.server + .inner + .build_server() + .core + .spam + .pyzor + .clone() + .expect("Pyzor enabled") +}