Settings reload: no DNS at build time, don't refuse over old failures #38
Vendored
+80
-29
@@ -20,13 +20,20 @@ use ahash::AHashMap;
|
|||||||
use directory::Directories;
|
use directory::Directories;
|
||||||
use registry::{
|
use registry::{
|
||||||
schema::{prelude::ObjectType, structs::BlockedIp},
|
schema::{prelude::ObjectType, structs::BlockedIp},
|
||||||
types::error::{Error, Warning},
|
types::{
|
||||||
|
error::{Error, Warning},
|
||||||
|
id::ObjectId,
|
||||||
|
},
|
||||||
};
|
};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use store::{LookupStores, registry::bootstrap::Bootstrap, write::now};
|
use store::{LookupStores, registry::bootstrap::Bootstrap, write::now};
|
||||||
|
|
||||||
pub struct ReloadResult {
|
pub struct ReloadResult {
|
||||||
|
/// Errors that kept the reload from being applied.
|
||||||
pub errors: Vec<Error>,
|
pub errors: Vec<Error>,
|
||||||
|
/// 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<Error>,
|
||||||
pub warnings: Vec<Warning>,
|
pub warnings: Vec<Warning>,
|
||||||
pub replaced_core: bool,
|
pub replaced_core: bool,
|
||||||
}
|
}
|
||||||
@@ -114,42 +121,60 @@ impl Server {
|
|||||||
directories: directory.directories,
|
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 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() {
|
if !self.has_new_build_errors(&bootstrap.errors) {
|
||||||
let core = Box::pin(Core::parse(&mut bootstrap, storage)).await;
|
servers
|
||||||
|
.parse_tcp_acceptors(&mut bootstrap, self.inner.clone())
|
||||||
|
.await;
|
||||||
|
|
||||||
if bootstrap.errors.is_empty() {
|
if !self.has_new_build_errors(&bootstrap.errors) {
|
||||||
let mut servers = Listeners::parse(&mut bootstrap).await;
|
// Update core
|
||||||
servers
|
self.inner.shared_core.store(core.into());
|
||||||
.parse_tcp_acceptors(&mut bootstrap, self.inner.clone())
|
|
||||||
.await;
|
|
||||||
|
|
||||||
if bootstrap.errors.is_empty() {
|
// Update tracers
|
||||||
// Update core
|
tracers.update();
|
||||||
self.inner.shared_core.store(core.into());
|
|
||||||
|
|
||||||
// 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
|
return Ok(ReloadResult {
|
||||||
self.inner
|
errors: Vec::new(),
|
||||||
.ipc
|
known_errors: bootstrap.errors,
|
||||||
.queue_tx
|
warnings: bootstrap.warnings,
|
||||||
.send(QueueEvent::ReloadSettings)
|
replaced_core: true,
|
||||||
.await
|
});
|
||||||
.ok();
|
|
||||||
|
|
||||||
return Ok(ReloadResult {
|
|
||||||
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) {
|
pub fn log(&self) {
|
||||||
for error in &self.errors {
|
for error in self.errors.iter().chain(&self.known_errors) {
|
||||||
error.log();
|
error.log();
|
||||||
}
|
}
|
||||||
for warning in &self.warnings {
|
for warning in &self.warnings {
|
||||||
@@ -176,8 +201,34 @@ impl From<Bootstrap> for ReloadResult {
|
|||||||
fn from(bootstrap: Bootstrap) -> Self {
|
fn from(bootstrap: Bootstrap) -> Self {
|
||||||
Self {
|
Self {
|
||||||
errors: bootstrap.errors,
|
errors: bootstrap.errors,
|
||||||
|
known_errors: Vec::new(),
|
||||||
warnings: bootstrap.warnings,
|
warnings: bootstrap.warnings,
|
||||||
replaced_core: false,
|
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<ObjectId> {
|
||||||
|
match error {
|
||||||
|
Error::Validation { object_id, .. }
|
||||||
|
| Error::Build { object_id, .. }
|
||||||
|
| Error::NotFound { object_id } => Some(*object_id),
|
||||||
|
Error::Internal { object_id, .. } => *object_id,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -96,6 +96,7 @@ impl Data {
|
|||||||
applications,
|
applications,
|
||||||
logos: Default::default(),
|
logos: Default::default(),
|
||||||
smtp_connectors: TlsConnectors::try_new().failed("Failed to build TLS connectors"),
|
smtp_connectors: TlsConnectors::try_new().failed("Failed to build TLS connectors"),
|
||||||
|
build_errors: Default::default(),
|
||||||
asn_geo_data: Default::default(),
|
asn_geo_data: Default::default(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -237,6 +238,7 @@ impl Default for Data {
|
|||||||
applications: WebApplications::new(),
|
applications: WebApplications::new(),
|
||||||
logos: Default::default(),
|
logos: Default::default(),
|
||||||
smtp_connectors: TlsConnectors::try_new().unwrap(),
|
smtp_connectors: TlsConnectors::try_new().unwrap(),
|
||||||
|
build_errors: Default::default(),
|
||||||
asn_geo_data: Default::default(),
|
asn_geo_data: Default::default(),
|
||||||
lookup_stores: Default::default(),
|
lookup_stores: Default::default(),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -16,7 +16,6 @@ use mail_auth::common::resolver::ToReverseName;
|
|||||||
use nlp::classifier::model::{CcfhClassifier, FhClassifier};
|
use nlp::classifier::model::{CcfhClassifier, FhClassifier};
|
||||||
use registry::schema::{
|
use registry::schema::{
|
||||||
enums::{ExpressionVariable, ModelSize},
|
enums::{ExpressionVariable, ModelSize},
|
||||||
prelude::ObjectType,
|
|
||||||
structs::{
|
structs::{
|
||||||
self, SpamDnsblServer, SpamDnsblSettings, SpamFileExtension, SpamPyzor, SpamRule,
|
self, SpamDnsblServer, SpamDnsblSettings, SpamFileExtension, SpamPyzor, SpamRule,
|
||||||
SpamSettings, SpamTag,
|
SpamSettings, SpamTag,
|
||||||
@@ -25,10 +24,10 @@ use registry::schema::{
|
|||||||
use sieve::SpamStatus;
|
use sieve::SpamStatus;
|
||||||
use std::{
|
use std::{
|
||||||
net::{IpAddr, SocketAddr},
|
net::{IpAddr, SocketAddr},
|
||||||
time::Duration,
|
sync::Arc,
|
||||||
|
time::{Duration, Instant},
|
||||||
};
|
};
|
||||||
use store::registry::{RegistryObject, bootstrap::Bootstrap};
|
use store::registry::{RegistryObject, bootstrap::Bootstrap};
|
||||||
use tokio::net::lookup_host;
|
|
||||||
use utils::{cache::CacheItemWeight, glob::GlobMap};
|
use utils::{cache::CacheItemWeight, glob::GlobMap};
|
||||||
|
|
||||||
#[derive(rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default)]
|
#[derive(rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Default)]
|
||||||
@@ -157,7 +156,11 @@ pub struct FtrlParameters {
|
|||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct PyzorConfig {
|
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<parking_lot::Mutex<Option<(SocketAddr, Instant)>>>,
|
||||||
pub timeout: Duration,
|
pub timeout: Duration,
|
||||||
pub min_count: u64,
|
pub min_count: u64,
|
||||||
pub min_wl_count: u64,
|
pub min_wl_count: u64,
|
||||||
@@ -474,31 +477,15 @@ impl PyzorConfig {
|
|||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
|
|
||||||
let port = pyzor.port;
|
// inbuxa: upstream resolved the host here and reported a failed lookup
|
||||||
let host = pyzor.host;
|
// as a build error, so a DNS hiccup on one node refused every settings
|
||||||
let address = match lookup_host(format!("{host}:{port}"))
|
// reload on it (and, from the node that ran ReloadSettings, across the
|
||||||
.await
|
// cluster). The lookup now happens when a message is checked; a
|
||||||
.map(|mut a| a.next())
|
// failure there is logged as a Pyzor error for that message.
|
||||||
{
|
|
||||||
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;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
PyzorConfig {
|
PyzorConfig {
|
||||||
address,
|
host: pyzor.host,
|
||||||
|
port: pyzor.port as u16,
|
||||||
|
resolved: Default::default(),
|
||||||
timeout: pyzor.timeout.into_inner(),
|
timeout: pyzor.timeout.into_inner(),
|
||||||
min_count: pyzor.block_count,
|
min_count: pyzor.block_count,
|
||||||
min_wl_count: pyzor.allow_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<SocketAddr> {
|
||||||
|
if let Ok(ip) = self.host.parse::<IpAddr>() {
|
||||||
|
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 {
|
impl ClassifierConfig {
|
||||||
pub async fn parse(bp: &mut Bootstrap) -> Option<Self> {
|
pub async fn parse(bp: &mut Bootstrap) -> Option<Self> {
|
||||||
let classifier = bp.setting_infallible::<structs::SpamClassifier>().await;
|
let classifier = bp.setting_infallible::<structs::SpamClassifier>().await;
|
||||||
|
|||||||
@@ -2,6 +2,8 @@
|
|||||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||||
*
|
*
|
||||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||||
|
*
|
||||||
|
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
use self::resolver::Policy;
|
use self::resolver::Policy;
|
||||||
@@ -22,7 +24,7 @@ use registry::schema::{
|
|||||||
};
|
};
|
||||||
use smtp_proto::*;
|
use smtp_proto::*;
|
||||||
use std::{
|
use std::{
|
||||||
net::{SocketAddr, ToSocketAddrs},
|
net::{IpAddr, SocketAddr},
|
||||||
str::FromStr,
|
str::FromStr,
|
||||||
time::Duration,
|
time::Duration,
|
||||||
};
|
};
|
||||||
@@ -384,19 +386,16 @@ impl SessionConfig {
|
|||||||
Some(Milter {
|
Some(Milter {
|
||||||
enable: bp.compile_expr(id, &milter.ctx_enable()),
|
enable: bp.compile_expr(id, &milter.ctx_enable()),
|
||||||
id,
|
id,
|
||||||
addrs: format!("{}:{}", milter.hostname, milter.port)
|
// inbuxa: upstream resolved the hostname here (a
|
||||||
.to_socket_addrs()
|
// blocking lookup) and made a failure a build error,
|
||||||
.map_err(|err| {
|
// which refused the whole settings reload. An IP
|
||||||
bp.build_error(
|
// address is kept as is; a name is resolved on each
|
||||||
id,
|
// connection (MilterClient::connect).
|
||||||
format!(
|
addrs: milter
|
||||||
"Unable to resolve milter hostname {}: {}",
|
.hostname
|
||||||
milter.hostname, err
|
.parse::<IpAddr>()
|
||||||
),
|
.map(|ip| vec![SocketAddr::new(ip, milter.port as u16)])
|
||||||
)
|
.unwrap_or_default(),
|
||||||
})
|
|
||||||
.ok()?
|
|
||||||
.collect(),
|
|
||||||
hostname: milter.hostname,
|
hostname: milter.hostname,
|
||||||
port: milter.port as u16,
|
port: milter.port as u16,
|
||||||
timeout_connect: milter.timeout_connect.into_inner(),
|
timeout_connect: milter.timeout_connect.into_inner(),
|
||||||
|
|||||||
@@ -166,6 +166,10 @@ pub struct Data {
|
|||||||
pub logos: Mutex<AHashMap<Box<str>, LogoCache>>,
|
pub logos: Mutex<AHashMap<Box<str>, LogoCache>>,
|
||||||
|
|
||||||
pub smtp_connectors: TlsConnectors,
|
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<AHashSet<registry::types::id::ObjectId>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
|
|||||||
@@ -240,6 +240,9 @@ impl BootManager {
|
|||||||
.parse_tcp_acceptors(&mut bootstrap, inner.clone())
|
.parse_tcp_acceptors(&mut bootstrap, inner.clone())
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
|
// inbuxa: a reload isn't refused over objects that failed here
|
||||||
|
inner.build_server().record_build_errors(&bootstrap.errors);
|
||||||
|
|
||||||
BootManager {
|
BootManager {
|
||||||
inner,
|
inner,
|
||||||
bootstrap,
|
bootstrap,
|
||||||
|
|||||||
@@ -2,6 +2,8 @@
|
|||||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||||
*
|
*
|
||||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
* 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};
|
use crate::registry::mapping::{RegistrySetResponse, map_bootstrap_error};
|
||||||
@@ -99,7 +101,7 @@ pub(crate) async fn action_set(
|
|||||||
} else {
|
} else {
|
||||||
set.response
|
set.response
|
||||||
.not_created
|
.not_created
|
||||||
.append(id, map_bootstrap_error(result.errors));
|
.append(id, reload_refused(result.errors));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Action::InvalidateCaches => {
|
Action::InvalidateCaches => {
|
||||||
@@ -573,3 +575,34 @@ async fn dmarc_troubleshoot(
|
|||||||
|
|
||||||
Some(request)
|
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<registry::types::error::Error>) -> SetError<Property> {
|
||||||
|
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::<Vec<_>>()
|
||||||
|
.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)
|
||||||
|
}
|
||||||
|
|||||||
@@ -947,10 +947,14 @@ impl RegistrySet for Server {
|
|||||||
self.cluster_broadcast(common::ipc::BroadcastEvent::RegistryChange(change))
|
self.cluster_broadcast(common::ipc::BroadcastEvent::RegistryChange(change))
|
||||||
.await;
|
.await;
|
||||||
}
|
}
|
||||||
Ok(_) => trc::event!(
|
Ok(reload) => {
|
||||||
Registry(trc::RegistryEvent::BuildWarning),
|
// inbuxa: name what stopped it
|
||||||
Details = "Settings didn't reload after a directory change",
|
reload.log();
|
||||||
),
|
trc::event!(
|
||||||
|
Registry(trc::RegistryEvent::BuildWarning),
|
||||||
|
Details = "Settings didn't reload after a directory change",
|
||||||
|
)
|
||||||
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
trc::error!(err.details("Failed to reload directories"));
|
trc::error!(err.details("Failed to reload directories"));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,8 @@
|
|||||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||||
*
|
*
|
||||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||||
|
*
|
||||||
|
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
use common::config::smtp::session::Milter;
|
use common::config::smtp::session::Milter;
|
||||||
@@ -25,7 +27,19 @@ impl MilterClient<TcpStream> {
|
|||||||
pub async fn connect(config: &Milter, session_id: u64) -> Result<Self> {
|
pub async fn connect(config: &Milter, session_id: u64) -> Result<Self> {
|
||||||
tokio::time::timeout(config.timeout_command, async {
|
tokio::time::timeout(config.timeout_command, async {
|
||||||
let mut last_err = Error::Disconnected;
|
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::<Vec<_>>();
|
||||||
|
&resolved
|
||||||
|
} else {
|
||||||
|
&config.addrs
|
||||||
|
};
|
||||||
|
for addr in addrs {
|
||||||
match TcpStream::connect(addr).await {
|
match TcpStream::connect(addr).await {
|
||||||
Ok(stream) => {
|
Ok(stream) => {
|
||||||
return Ok(MilterClient {
|
return Ok(MilterClient {
|
||||||
|
|||||||
@@ -48,19 +48,24 @@ pub(crate) async fn pyzor_check(
|
|||||||
// Send message to address. inbuxa: in tests, a fixed table answers
|
// Send message to address. inbuxa: in tests, a fixed table answers
|
||||||
// instead of a public server (test_response).
|
// instead of a public server (test_response).
|
||||||
#[cfg(not(feature = "test_mode"))]
|
#[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")]
|
#[cfg(feature = "test_mode")]
|
||||||
let response = std::io::Result::Ok(test_response(&request));
|
let response = std::io::Result::Ok(test_response(&request));
|
||||||
|
|
||||||
response
|
response.map(Into::into).map_err(|err| {
|
||||||
.map(Into::into)
|
trc::SpamEvent::PyzorError
|
||||||
.map_err(|err| {
|
.into_err()
|
||||||
trc::SpamEvent::PyzorError
|
.ctx(trc::Key::Url, format!("{}:{}", config.host, config.port))
|
||||||
.into_err()
|
.reason(err)
|
||||||
.ctx(trc::Key::Url, config.address.to_string())
|
.details("Pyzor failed")
|
||||||
.reason(err)
|
})
|
||||||
.details("Pyzor failed")
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// inbuxa: the answers tests get, by digest, instead of a public server's,
|
/// inbuxa: the answers tests get, by digest, instead of a public server's,
|
||||||
|
|||||||
@@ -2,6 +2,8 @@
|
|||||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||||
*
|
*
|
||||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||||
|
*
|
||||||
|
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
@@ -75,7 +77,7 @@ async fn milter_session() {
|
|||||||
else_: "true".into(),
|
else_: "true".into(),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
},
|
},
|
||||||
hostname: "127.0.0.1".into(),
|
hostname: "localhost".into(), // inbuxa: resolved when the session connects
|
||||||
port: 9332,
|
port: 9332,
|
||||||
use_tls: false,
|
use_tls: false,
|
||||||
stages: Map::new(vec![MtaStage::Data]),
|
stages: Map::new(vec![MtaStage::Data]),
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ pub mod monitoring;
|
|||||||
pub mod oidc;
|
pub mod oidc;
|
||||||
pub mod purge;
|
pub mod purge;
|
||||||
pub mod quota;
|
pub mod quota;
|
||||||
|
pub mod reload; // inbuxa: reloads and build errors
|
||||||
pub mod security;
|
pub mod security;
|
||||||
pub mod task;
|
pub mod task;
|
||||||
pub mod tenant;
|
pub mod tenant;
|
||||||
|
|||||||
@@ -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",
|
||||||
|
"[email protected]",
|
||||||
|
"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("[email protected]");
|
||||||
|
|
||||||
|
// 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("[email protected]");
|
||||||
|
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")
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user