Compare commits
3
Commits
f55087dd9b
...
999ae12cc7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
999ae12cc7 | ||
|
|
5853831bad | ||
|
|
639a415a4f |
Vendored
+80
-29
@@ -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<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 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<Bootstrap> 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<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,
|
||||
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(),
|
||||
}
|
||||
|
||||
@@ -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<parking_lot::Mutex<Option<(SocketAddr, Instant)>>>,
|
||||
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<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 {
|
||||
pub async fn parse(bp: &mut Bootstrap) -> Option<Self> {
|
||||
let classifier = bp.setting_infallible::<structs::SpamClassifier>().await;
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* 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::<IpAddr>()
|
||||
.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(),
|
||||
|
||||
@@ -166,6 +166,10 @@ pub struct Data {
|
||||
pub logos: Mutex<AHashMap<Box<str>, 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<AHashSet<registry::types::id::ObjectId>>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* 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<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))
|
||||
.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"));
|
||||
}
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* 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<TcpStream> {
|
||||
pub async fn connect(config: &Milter, session_id: u64) -> Result<Self> {
|
||||
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::<Vec<_>>();
|
||||
&resolved
|
||||
} else {
|
||||
&config.addrs
|
||||
};
|
||||
for addr in addrs {
|
||||
match TcpStream::connect(addr).await {
|
||||
Ok(stream) => {
|
||||
return Ok(MilterClient {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::{
|
||||
@@ -19,7 +21,7 @@ use crate::{
|
||||
write::SearchIndex,
|
||||
};
|
||||
use mysql_async::{IsolationLevel, TxOpts, Value, prelude::Queryable};
|
||||
use nlp::tokenizers::word::WordTokenizer;
|
||||
use nlp::{language::Language, tokenizers::word::WordTokenizer};
|
||||
use std::fmt::Write;
|
||||
|
||||
impl MysqlStore {
|
||||
@@ -146,6 +148,20 @@ impl MysqlStore {
|
||||
}
|
||||
}
|
||||
|
||||
// inbuxa: InnoDB's default full-text stopword list
|
||||
// (INFORMATION_SCHEMA.INNODB_FT_DEFAULT_STOPWORD) and innodb_ft_min_token_size
|
||||
// default; words outside these are not in a FULLTEXT index.
|
||||
const FT_STOPWORDS: &[&str] = &[
|
||||
"a", "about", "an", "are", "as", "at", "be", "by", "com", "de", "en", "for", "from", "how",
|
||||
"i", "in", "is", "it", "la", "of", "on", "or", "that", "the", "this", "to", "was", "what",
|
||||
"when", "where", "who", "will", "with", "und", "www",
|
||||
];
|
||||
const FT_MIN_TOKEN_SIZE: usize = 3;
|
||||
|
||||
fn is_ft_indexed(word: &str) -> bool {
|
||||
word.chars().count() >= FT_MIN_TOKEN_SIZE && !FT_STOPWORDS.contains(&word)
|
||||
}
|
||||
|
||||
fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
|
||||
if filters.is_empty() {
|
||||
return Vec::new();
|
||||
@@ -171,30 +187,77 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
|
||||
|
||||
if field.is_text() && matches!(op, SearchOperator::Equal | SearchOperator::Contains)
|
||||
{
|
||||
let (value, mode) = match (value, op) {
|
||||
(SearchValue::Text { value, .. }, SearchOperator::Equal) => {
|
||||
(Value::Bytes(format!("{value:?}").into_bytes()), "BOOLEAN")
|
||||
}
|
||||
(SearchValue::Text { value, .. }, ..) => {
|
||||
let (value, mode, unindexed) = match (value, op) {
|
||||
(SearchValue::Text { value, .. }, SearchOperator::Equal) => (
|
||||
Value::Bytes(format!("{value:?}").into_bytes()),
|
||||
"BOOLEAN",
|
||||
Vec::new(),
|
||||
),
|
||||
(SearchValue::Text { value, language }, ..) => {
|
||||
let mut text_query = String::with_capacity(value.len() + 1);
|
||||
let mut unindexed = Vec::new();
|
||||
|
||||
for item in WordTokenizer::new(value, MAX_TOKEN_LENGTH) {
|
||||
if !text_query.is_empty() {
|
||||
text_query.push(' ');
|
||||
// inbuxa: InnoDB never indexes stopwords ("com",
|
||||
// "de", "www", ...) or words under
|
||||
// innodb_ft_min_token_size, and a required
|
||||
// (+word) term it has not indexed matches no row,
|
||||
// so "example.com" or "[email protected]" found
|
||||
// nothing. Such words are matched with a
|
||||
// word-boundary REGEXP instead.
|
||||
if is_ft_indexed(&item.word) {
|
||||
if !text_query.is_empty() {
|
||||
text_query.push(' ');
|
||||
}
|
||||
text_query.push('+');
|
||||
text_query.push_str(&item.word);
|
||||
} else {
|
||||
unindexed.push(item.word);
|
||||
}
|
||||
text_query.push('+');
|
||||
text_query.push_str(&item.word);
|
||||
}
|
||||
|
||||
(Value::Bytes(text_query.into_bytes()), "BOOLEAN")
|
||||
// For language text (bodies, subjects) the unindexed
|
||||
// words are noise words and only checked when nothing
|
||||
// else is left to match; keyword text (addresses,
|
||||
// contact fields) checks every word, as the other
|
||||
// backends do.
|
||||
if !text_query.is_empty() && !matches!(language, Language::None) {
|
||||
unindexed.clear();
|
||||
}
|
||||
|
||||
(Value::Bytes(text_query.into_bytes()), "BOOLEAN", unindexed)
|
||||
}
|
||||
_ => {
|
||||
debug_assert!(false, "Invalid search value for text field");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let _ = write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column());
|
||||
values.push(value);
|
||||
if unindexed.is_empty() {
|
||||
let _ =
|
||||
write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column());
|
||||
values.push(value);
|
||||
} else {
|
||||
query.push('(');
|
||||
let is_empty = matches!(&value, Value::Bytes(v) if v.is_empty());
|
||||
if !is_empty {
|
||||
let _ = write!(
|
||||
query,
|
||||
"MATCH({}) AGAINST(? IN {mode} MODE) AND ",
|
||||
field.column()
|
||||
);
|
||||
values.push(value);
|
||||
}
|
||||
for (i, word) in unindexed.iter().enumerate() {
|
||||
if i > 0 {
|
||||
query.push_str(" AND ");
|
||||
}
|
||||
let _ = write!(query, "{} REGEXP ?", field.column());
|
||||
values.push(Value::Bytes(
|
||||
format!("(^|[^[:alnum:]]){word}([^[:alnum:]]|$)").into_bytes(),
|
||||
));
|
||||
}
|
||||
query.push(')');
|
||||
}
|
||||
} else if let SearchValue::KeyValues(kv) = value {
|
||||
let (key, value) = kv.iter().next().unwrap();
|
||||
|
||||
|
||||
@@ -2,12 +2,17 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::{
|
||||
backend::postgres::{
|
||||
DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error,
|
||||
into_pool_error, is_timeout_error,
|
||||
backend::{
|
||||
MAX_TOKEN_LENGTH,
|
||||
postgres::{
|
||||
DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error,
|
||||
into_pool_error, is_timeout_error,
|
||||
},
|
||||
},
|
||||
search::{
|
||||
IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchOperator,
|
||||
@@ -15,7 +20,7 @@ use crate::{
|
||||
},
|
||||
write::SearchIndex,
|
||||
};
|
||||
use nlp::language::Language;
|
||||
use nlp::{language::Language, tokenizers::space::SpaceTokenizer};
|
||||
use std::fmt::Write;
|
||||
use tokio_postgres::{
|
||||
IsolationLevel,
|
||||
@@ -43,6 +48,19 @@ impl PostgresStore {
|
||||
let primary_keys = index.primary_keys();
|
||||
let all_fields = index.all_fields();
|
||||
let fields = document.fields;
|
||||
// inbuxa: keyword text (addresses, contact fields, ...) is split into
|
||||
// words before it reaches the text parser, see keyword_terms().
|
||||
let keywords = primary_keys
|
||||
.iter()
|
||||
.chain(all_fields)
|
||||
.map(|field| match fields.get(field) {
|
||||
Some(SearchValue::Text {
|
||||
value,
|
||||
language: Language::None,
|
||||
}) if field.is_text() => Some(keyword_terms(value)),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let mut values = Vec::with_capacity(fields.len() + 2);
|
||||
let mut query = format!("INSERT INTO {} (", index.psql_table());
|
||||
|
||||
@@ -74,7 +92,20 @@ impl PostgresStore {
|
||||
(0, PG_UNSTEMMED_LANG)
|
||||
};
|
||||
|
||||
if field.is_text() {
|
||||
if let Some(keywords) = &keywords[i] {
|
||||
let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})");
|
||||
values.push(keywords as &(dyn ToSql + Sync));
|
||||
if field.sort_column().is_some() {
|
||||
let value_ref = format!("${}", values.len() + 1);
|
||||
if text_len > 255 {
|
||||
let _ = write!(&mut query, ",left({value_ref},255)");
|
||||
} else {
|
||||
let _ = write!(&mut query, ",{value_ref}");
|
||||
}
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
}
|
||||
continue;
|
||||
} else if field.is_text() {
|
||||
let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})");
|
||||
} else if text_len > 512 {
|
||||
query.push_str("left(");
|
||||
@@ -134,6 +165,7 @@ impl PostgresStore {
|
||||
) -> trc::Result<Vec<R>> {
|
||||
let mut query = format!("SELECT {} FROM {}", R::field().column(), index.psql_table());
|
||||
let params = self.build_filter(&mut query, filters);
|
||||
let params = params.iter().map(SqlParam::as_sql).collect::<Vec<_>>();
|
||||
if !sort.is_empty() {
|
||||
build_sort(&mut query, sort);
|
||||
}
|
||||
@@ -155,6 +187,7 @@ impl PostgresStore {
|
||||
let table = filter.index.psql_table();
|
||||
let mut where_clause = String::new();
|
||||
let params = self.build_filter(&mut where_clause, &filter.filters);
|
||||
let params = params.iter().map(SqlParam::as_sql).collect::<Vec<_>>();
|
||||
let conn = self.conn_pool.get().await.map_err(into_pool_error)?;
|
||||
let s = conn
|
||||
.prepare_cached(&format!("DELETE FROM {table}{where_clause}"))
|
||||
@@ -196,7 +229,7 @@ impl PostgresStore {
|
||||
&self,
|
||||
query: &mut String,
|
||||
filters: &'x [SearchFilter],
|
||||
) -> Vec<&'x (dyn ToSql + Sync)> {
|
||||
) -> Vec<SqlParam<'x>> {
|
||||
if filters.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
@@ -237,6 +270,10 @@ impl PostgresStore {
|
||||
|
||||
if matches!(language, Language::None) {
|
||||
let _ = write!(query, "@@ {method}('{config}', ${value_pos})");
|
||||
if let SearchValue::Text { value, .. } = value {
|
||||
values.push(SqlParam::Owned(keyword_terms(value)));
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
let _ = write!(query, "@@ ({method}('{config}', ${value_pos})");
|
||||
for fallback in [PG_FALLBACK_LANG, PG_UNSTEMMED_LANG] {
|
||||
@@ -247,18 +284,18 @@ impl PostgresStore {
|
||||
}
|
||||
query.push(')');
|
||||
}
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
values.push(SqlParam::Ref(value));
|
||||
} else if let SearchValue::KeyValues(kv) = value {
|
||||
query.push_str(field.column());
|
||||
query.push(' ');
|
||||
|
||||
let (key, value) = kv.iter().next().unwrap();
|
||||
values.push(key as &(dyn ToSql + Sync));
|
||||
values.push(SqlParam::Ref(key));
|
||||
|
||||
if !value.is_empty() {
|
||||
let _ = write!(query, "->> ${value_pos} ");
|
||||
op.write_pqsql(query, values.len() + 1);
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
values.push(SqlParam::Ref(value));
|
||||
} else {
|
||||
let _ = write!(query, " ? ${value_pos}");
|
||||
}
|
||||
@@ -267,7 +304,7 @@ impl PostgresStore {
|
||||
query.push(' ');
|
||||
|
||||
op.write_pqsql(query, value_pos);
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
values.push(SqlParam::Ref(value));
|
||||
}
|
||||
}
|
||||
SearchFilter::And | SearchFilter::Or => {
|
||||
@@ -321,6 +358,38 @@ impl PostgresStore {
|
||||
}
|
||||
}
|
||||
|
||||
// inbuxa: PostgreSQL's text parser keeps "[email protected]" (and host names,
|
||||
// URLs, file paths, ...) as a single token, so a search for "user" or
|
||||
// "example.com" never matched an address. Keyword text is split into words the
|
||||
// same way the built-in index splits it (SpaceTokenizer: lowercase runs of
|
||||
// alphanumerics) on both the indexing and the query side, so a full address,
|
||||
// its local part, its domain and the display-name words all match, as they do
|
||||
// on the other backends.
|
||||
pub(crate) fn keyword_terms(value: &str) -> String {
|
||||
let mut terms = String::with_capacity(value.len());
|
||||
for token in SpaceTokenizer::new(value, MAX_TOKEN_LENGTH) {
|
||||
if !terms.is_empty() {
|
||||
terms.push(' ');
|
||||
}
|
||||
terms.push_str(&token);
|
||||
}
|
||||
terms
|
||||
}
|
||||
|
||||
pub(super) enum SqlParam<'x> {
|
||||
Ref(&'x (dyn ToSql + Sync)),
|
||||
Owned(String),
|
||||
}
|
||||
|
||||
impl SqlParam<'_> {
|
||||
fn as_sql(&self) -> &(dyn ToSql + Sync) {
|
||||
match self {
|
||||
SqlParam::Ref(value) => *value,
|
||||
SqlParam::Owned(value) => value,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn build_sort(query: &mut String, sort: &[SearchComparator]) {
|
||||
query.push_str(" ORDER BY ");
|
||||
for (i, comparator) in sort.iter().enumerate() {
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* 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]),
|
||||
|
||||
@@ -128,6 +128,11 @@ pub async fn test(test: &TestServer) {
|
||||
println!("Running trace document tests...");
|
||||
test_trace_documents(store.clone()).await;
|
||||
|
||||
// inbuxa: address fields match by full address, local part, domain and
|
||||
// display name on every backend
|
||||
println!("Running address search tests...");
|
||||
test_address_search(store.clone()).await;
|
||||
|
||||
// Large document insert test
|
||||
println!("Running large document insert tests...");
|
||||
let mut large_text = String::with_capacity(20 * 1024 * 1024);
|
||||
@@ -972,3 +977,155 @@ async fn test_trace_documents(store: SearchStore) {
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
// inbuxa: the message indexer passes each display name and each address of
|
||||
// From/To/Cc/Bcc as keyword text (Language::None). The built-in index splits
|
||||
// that text into words, so an address is found by its full form, its local
|
||||
// part, its domain or a display-name word; PostgreSQL kept the whole address
|
||||
// as one token and MySQL dropped stopwords such as "com" and words under three
|
||||
// characters. The expected results below are the built-in (RocksDB/SQLite)
|
||||
// results and must be the same on every backend.
|
||||
async fn test_address_search(store: SearchStore) {
|
||||
const ACCOUNT_ID: u32 = 7;
|
||||
let messages: [[&[(&str, &str)]; 4]; 5] = [
|
||||
// From, To, Cc, Bcc
|
||||
[
|
||||
&[("Amazon.com", "[email protected]")],
|
||||
&[("Jane Doe", "[email protected]")],
|
||||
&[],
|
||||
&[],
|
||||
],
|
||||
[
|
||||
&[("", "[email protected]")],
|
||||
&[("", "[email protected]")],
|
||||
&[("Jane Doe", "[email protected]")],
|
||||
&[],
|
||||
],
|
||||
[
|
||||
&[("GitHub", "[email protected]")],
|
||||
&[("Jo Li", "[email protected]")],
|
||||
&[],
|
||||
&[("Audit", "[email protected]")],
|
||||
],
|
||||
[
|
||||
&[("Jane Doe", "[email protected]")],
|
||||
&[("Amazon Web Services", "[email protected]")],
|
||||
&[("Bob", "[email protected]")],
|
||||
&[("", "[email protected]")],
|
||||
],
|
||||
[
|
||||
&[("Newsletter", "[email protected]")],
|
||||
&[("", "[email protected]")],
|
||||
&[],
|
||||
&[],
|
||||
],
|
||||
];
|
||||
let fields = [
|
||||
EmailSearchField::From,
|
||||
EmailSearchField::To,
|
||||
EmailSearchField::Cc,
|
||||
EmailSearchField::Bcc,
|
||||
];
|
||||
|
||||
let mut documents = Vec::new();
|
||||
let mut mask = RoaringBitmap::new();
|
||||
for (document_id, message) in messages.iter().enumerate() {
|
||||
let mut document = IndexDocument::new(SearchIndex::Email)
|
||||
.with_account_id(ACCOUNT_ID)
|
||||
.with_document_id(document_id as u32);
|
||||
for (field, addresses) in fields.iter().zip(message.iter()) {
|
||||
for (name, address) in addresses.iter() {
|
||||
if !name.is_empty() {
|
||||
document.index_text(field.clone(), name, Language::None);
|
||||
}
|
||||
document.index_text(field.clone(), address, Language::None);
|
||||
}
|
||||
}
|
||||
document.index_unsigned(EmailSearchField::ReceivedAt, document_id as u64);
|
||||
documents.push(document);
|
||||
mask.insert(document_id as u32);
|
||||
}
|
||||
store.index(documents).await.unwrap();
|
||||
if let SearchStore::ElasticSearch(store) = &store {
|
||||
store.refresh_index(SearchIndex::Email).await.unwrap();
|
||||
}
|
||||
|
||||
for (field, text, expected) in [
|
||||
// full address
|
||||
(EmailSearchField::From, "[email protected]", vec![0u32]),
|
||||
(EmailSearchField::To, "[email protected]", vec![0]),
|
||||
(EmailSearchField::Cc, "[email protected]", vec![1]),
|
||||
(EmailSearchField::Bcc, "[email protected]", vec![3]),
|
||||
(EmailSearchField::To, "[email protected]", vec![1, 2]),
|
||||
// local part
|
||||
(EmailSearchField::From, "noreply", vec![0, 2]),
|
||||
(EmailSearchField::To, "jo", vec![1, 2]),
|
||||
(EmailSearchField::Cc, "bob", vec![3]),
|
||||
(EmailSearchField::Bcc, "audit", vec![2]),
|
||||
// domain
|
||||
(EmailSearchField::From, "amazon.com", vec![0, 1]),
|
||||
(EmailSearchField::From, "amazon", vec![0, 1]),
|
||||
(EmailSearchField::To, "example.org", vec![0, 4]),
|
||||
(EmailSearchField::To, "io.de", vec![1, 2]),
|
||||
(EmailSearchField::Cc, "example.net", vec![3]),
|
||||
(EmailSearchField::Bcc, "example.org", vec![2]),
|
||||
(EmailSearchField::From, "www.example.com", vec![4]),
|
||||
(EmailSearchField::From, "com", vec![0, 1, 2, 4]),
|
||||
// display name
|
||||
(EmailSearchField::From, "Jane", vec![3]),
|
||||
(EmailSearchField::From, "jane doe", vec![3]),
|
||||
(EmailSearchField::To, "Web Services", vec![3]),
|
||||
(EmailSearchField::To, "Li", vec![2]),
|
||||
(EmailSearchField::Cc, "Doe", vec![1]),
|
||||
(EmailSearchField::Bcc, "Audit", vec![2]),
|
||||
// hyphenated local part
|
||||
(EmailSearchField::From, "shipment-tracking", vec![1]),
|
||||
(EmailSearchField::From, "tracking", vec![1]),
|
||||
// no match
|
||||
(EmailSearchField::From, "amazon.org", vec![]),
|
||||
(EmailSearchField::To, "noreply", vec![]),
|
||||
(EmailSearchField::Bcc, "jane", vec![]),
|
||||
] {
|
||||
let ids = store
|
||||
.query_account(
|
||||
SearchQuery::new(SearchIndex::Email)
|
||||
.with_filters(vec![
|
||||
SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID),
|
||||
SearchFilter::has_keyword(field.clone(), text),
|
||||
])
|
||||
.with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt))
|
||||
.with_mask(mask.clone()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(ids, expected, "{field:?} {text:?}");
|
||||
}
|
||||
|
||||
// TEXT-style search across all address fields
|
||||
let ids = store
|
||||
.query_account(
|
||||
SearchQuery::new(SearchIndex::Email)
|
||||
.with_filters(vec![
|
||||
SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID),
|
||||
SearchFilter::Or,
|
||||
SearchFilter::has_keyword(EmailSearchField::From, "example.org"),
|
||||
SearchFilter::has_keyword(EmailSearchField::To, "example.org"),
|
||||
SearchFilter::has_keyword(EmailSearchField::Cc, "example.org"),
|
||||
SearchFilter::has_keyword(EmailSearchField::Bcc, "example.org"),
|
||||
SearchFilter::End,
|
||||
])
|
||||
.with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt))
|
||||
.with_mask(mask.clone()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(ids, vec![0, 1, 2, 3, 4]);
|
||||
|
||||
store
|
||||
.unindex(
|
||||
SearchQuery::new(SearchIndex::Email)
|
||||
.with_filter(SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID)),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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