jcoffey-dev
999ae12cc7
A 3-node rehearsal found every settings reload refused, cluster-wide,
because one node couldn't resolve the Pyzor server:
- PyzorConfig::parse resolved the host while building the settings and
made a failed lookup a build error. It now keeps the host and port and
resolves when a message is checked (an IP address is used as is, a
name is reused for five minutes, the lookup counts against the Pyzor
timeout). A failure there is a Pyzor error for that message.
- A milter's hostname was resolved the same way, with a blocking
to_socket_addrs in async code. An IP address is kept; a name is now
resolved on each connection.
Other build-time I/O is already non-fatal: directories that can't
connect become unavailable with a warning (DIR-21), and the AI model
locality check only warns.
reload_registry swapped the core only when the whole build was free of
errors, while boot runs with whatever built. One failing object thus
refused every later reload, and the running settings went stale. Now a
reload is refused only for errors in objects that built when the
running settings were built (at boot or by the last applied reload):
applying it would lose those. Objects that already failed then are
missing from the running settings anyway, as at boot, so their errors
are logged and returned as known_errors but don't hold the reload back.
Refusing on new errors keeps a bad edit from taking a working object
out of service; the admin gets the error instead.
ReloadSettings now says "Settings were not reloaded." and names the
object and its error ("Tracer with id ...: Only one console tracer is
allowed"), with a count of any further errors. A refused reload after a
directory change logs its errors too.
system::reload::reload_tests (new): with Pyzor enabled on an
unresolvable host, ReloadSettings succeeds (on main it fails with
"Invalid address: failed to lookup address information"); an IP host
needs no lookup; a new build error refuses the reload, names the object
and leaves the running settings unchanged; the same error, once known
from the running settings' build, no longer blocks; once fixed, a new
error there blocks again. smtp::inbound::milter's session test now
names its milter "localhost", so the connect-time lookup is exercised.
314 lines
11 KiB
Rust
314 lines
11 KiB
Rust
/*
|
|
* 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 super::{backup::BackupParams, console::store_console};
|
|
use crate::{
|
|
BuildServer, Caches, Core, Data, IPC_CHANNEL_BUFFER, Inner, Ipc,
|
|
config::{
|
|
network::AsnGeoLookupConfig, server::Listeners, storage::Storage, telemetry::Telemetry,
|
|
},
|
|
ipc::{BroadcastEvent, PushEvent, QueueEvent, ReportingEvent, TrainTaskController},
|
|
manager::defaults::BootstrapDefaults,
|
|
};
|
|
use arc_swap::ArcSwap;
|
|
use std::{
|
|
net::{IpAddr, Ipv4Addr},
|
|
path::PathBuf,
|
|
sync::Arc,
|
|
};
|
|
use store::{RegistryStore, registry::bootstrap::Bootstrap};
|
|
use tokio::sync::{Notify, mpsc};
|
|
use utils::{UnwrapFailure, failed};
|
|
|
|
pub struct BootManager {
|
|
pub bootstrap: Bootstrap,
|
|
pub inner: Arc<Inner>,
|
|
pub servers: Listeners,
|
|
pub ipc_rxs: IpcReceivers,
|
|
}
|
|
|
|
pub struct IpcReceivers {
|
|
pub push_rx: Option<mpsc::Receiver<PushEvent>>,
|
|
pub queue_rx: Option<mpsc::Receiver<QueueEvent>>,
|
|
pub report_rx: Option<mpsc::Receiver<ReportingEvent>>,
|
|
pub broadcast_rx: Option<mpsc::Receiver<BroadcastEvent>>,
|
|
}
|
|
|
|
const HELP: &str = concat!(
|
|
types::brand_server!(),
|
|
" ",
|
|
types::brand_version_full!(),
|
|
r#"
|
|
|
|
Usage: inbuxa [OPTIONS]
|
|
|
|
Options:
|
|
-c, --config <PATH> Start server with the specified configuration file
|
|
-e, --export <PATH> Export all store data to a specific path
|
|
-i, --import <PATH> Import store data from a specific path
|
|
-o, --console Open the store console
|
|
-h, --help Print help
|
|
-V, --version Print version
|
|
|
|
An export holds everything in the data and blob stores except short-lived
|
|
in-memory state (rate limits, locks, greylisting) and the full-text search
|
|
index, which belongs to one search backend. An import into an empty store
|
|
queues the index to be rebuilt when the server next starts. EXPORT_TYPES
|
|
limits an export to some of: data, registry, blob, changelog, queue, report,
|
|
telemetry, tasks.
|
|
"#
|
|
);
|
|
|
|
#[derive(PartialEq, Eq)]
|
|
enum StoreOp {
|
|
Export(BackupParams),
|
|
Import(PathBuf),
|
|
Console,
|
|
None,
|
|
}
|
|
|
|
impl BootManager {
|
|
pub async fn init() -> Self {
|
|
let mut config_path = std::env::var("CONFIG_PATH").ok();
|
|
let mut import_export = StoreOp::None;
|
|
|
|
if config_path.is_none() {
|
|
let mut args = std::env::args().skip(1);
|
|
|
|
while let Some(arg) = args.next().and_then(|arg| {
|
|
arg.strip_prefix("--")
|
|
.or_else(|| arg.strip_prefix('-'))
|
|
.map(|arg| arg.to_string())
|
|
}) {
|
|
let (key, value) = if let Some((key, value)) = arg.split_once('=') {
|
|
(key.to_string(), Some(value.trim().to_string()))
|
|
} else {
|
|
(arg, args.next())
|
|
};
|
|
|
|
match (key.as_str(), value) {
|
|
("help" | "h", _) => {
|
|
eprintln!("{HELP}");
|
|
std::process::exit(0);
|
|
}
|
|
("version" | "V", _) => {
|
|
println!("{}", types::brand_version_full!());
|
|
std::process::exit(0);
|
|
}
|
|
("config" | "c", Some(value)) => {
|
|
config_path = Some(value);
|
|
}
|
|
("export" | "e", Some(value)) => {
|
|
import_export = StoreOp::Export(BackupParams::new(value.into()));
|
|
}
|
|
("import" | "i", Some(value)) => {
|
|
import_export = StoreOp::Import(value.into());
|
|
}
|
|
("console" | "o", None) => {
|
|
import_export = StoreOp::Console;
|
|
}
|
|
(_, None) => {
|
|
failed(&format!("Unrecognized command '{key}', try '--help'."));
|
|
}
|
|
(_, Some(_)) => failed(&format!(
|
|
"Missing value for argument '{key}', try '--help'."
|
|
)),
|
|
}
|
|
}
|
|
|
|
if config_path.is_none() {
|
|
if import_export == StoreOp::None {
|
|
eprintln!("{HELP}");
|
|
} else {
|
|
eprintln!("Missing '--config' argument for import/export.")
|
|
}
|
|
std::process::exit(0);
|
|
}
|
|
}
|
|
|
|
// Initialize registry
|
|
let registry = RegistryStore::init(
|
|
PathBuf::from(config_path.unwrap()),
|
|
import_export == StoreOp::None,
|
|
)
|
|
.await
|
|
.failed("⚠️ Startup failed");
|
|
let mut bootstrap = Bootstrap::new(registry).await;
|
|
|
|
// Add safe defaults if missing
|
|
if import_export == StoreOp::None {
|
|
bootstrap.insert_safe_defaults().await;
|
|
}
|
|
|
|
// Start listeners
|
|
let mut servers = Listeners::parse(&mut bootstrap).await;
|
|
servers.bind_and_drop_priv(&mut bootstrap);
|
|
|
|
// Parse storage
|
|
let storage = Storage::parse(&mut bootstrap).await;
|
|
|
|
// Parse telemetry
|
|
let telemetry = Telemetry::parse(&mut bootstrap, &storage).await;
|
|
|
|
match import_export {
|
|
StoreOp::None => {
|
|
// Parse components
|
|
let core: Box<Core> =
|
|
Box::new(Box::pin(Core::parse(&mut bootstrap, storage)).await);
|
|
let data = Data::parse(&mut bootstrap).await;
|
|
let cache = Caches::parse(&mut bootstrap).await;
|
|
|
|
// Enable telemetry
|
|
|
|
|
|
telemetry.enable();
|
|
|
|
if bootstrap.registry.is_bootstrap_mode() {
|
|
trc::event!(
|
|
Server(trc::ServerEvent::BootstrapMode),
|
|
Hostname = bootstrap.registry.local_hostname().to_string(),
|
|
Details =
|
|
"No configuration file was found. Port 8080 is open for initial setup.",
|
|
Version = types::brand_version_full!(),
|
|
);
|
|
} else if bootstrap.registry.is_recovery_mode() {
|
|
trc::event!(
|
|
Server(trc::ServerEvent::RecoveryMode),
|
|
Details = "Port 8080 is open for troubleshooting and recovery.",
|
|
Hostname = bootstrap.registry.local_hostname().to_string(),
|
|
Version = types::brand_version_full!(),
|
|
);
|
|
} else {
|
|
trc::event!(
|
|
Server(trc::ServerEvent::Startup),
|
|
Hostname = bootstrap.registry.local_hostname().to_string(),
|
|
Version = types::brand_version_full!(),
|
|
);
|
|
}
|
|
|
|
if core.storage.coordinator.is_enabled() {
|
|
trc::event!(
|
|
Cluster(trc::ClusterEvent::Startup),
|
|
Id = bootstrap.registry.node_id(),
|
|
Type = bootstrap
|
|
.registry
|
|
.cluster_role()
|
|
.unwrap_or("[default]")
|
|
.to_string(),
|
|
Details = bootstrap.registry.cluster_push_shard()
|
|
);
|
|
}
|
|
|
|
// Build shared inner
|
|
let has_remote_asn = matches!(
|
|
core.network.asn_geo_lookup,
|
|
AsnGeoLookupConfig::Resource { .. }
|
|
);
|
|
let (ipc, ipc_rxs) = build_ipc(!core.storage.coordinator.is_none());
|
|
let inner = Arc::new(Inner {
|
|
shared_core: ArcSwap::new(Arc::from(core)),
|
|
data,
|
|
ipc,
|
|
cache,
|
|
});
|
|
|
|
if !bootstrap.registry.is_recovery_mode() {
|
|
// Load spam model
|
|
if let Err(err) = inner.build_server().spam_model_reload().await {
|
|
trc::error!(
|
|
err.details("Failed to load spam filter model")
|
|
.caused_by(trc::location!())
|
|
);
|
|
}
|
|
|
|
// Fetch ASN database
|
|
if has_remote_asn {
|
|
inner
|
|
.build_server()
|
|
.lookup_asn_country(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))
|
|
.await;
|
|
}
|
|
}
|
|
|
|
// Parse TCP acceptors
|
|
servers
|
|
.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,
|
|
servers,
|
|
ipc_rxs,
|
|
}
|
|
}
|
|
StoreOp::Export(path) => {
|
|
// Enable telemetry
|
|
telemetry.enable();
|
|
|
|
// Parse settings and backup
|
|
Box::pin(Core::parse(&mut bootstrap, storage))
|
|
.await
|
|
.backup(path)
|
|
.await;
|
|
std::process::exit(0);
|
|
}
|
|
StoreOp::Import(path) => {
|
|
// Enable telemetry
|
|
telemetry.enable();
|
|
|
|
// Parse settings and restore
|
|
let core = Box::pin(Core::parse(&mut bootstrap, storage)).await;
|
|
let imported = core.restore(path).await;
|
|
// inbuxa: the search index isn't exported; rebuild it
|
|
core.queue_reindex(&imported).await;
|
|
std::process::exit(0);
|
|
}
|
|
StoreOp::Console => {
|
|
// Store console
|
|
store_console(
|
|
Box::pin(Core::parse(&mut bootstrap, storage))
|
|
.await
|
|
.storage
|
|
.data,
|
|
)
|
|
.await;
|
|
std::process::exit(0);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn build_ipc(has_pubsub: bool) -> (Ipc, IpcReceivers) {
|
|
// Build ipc receivers
|
|
let (push_tx, push_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
|
|
let (queue_tx, queue_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
|
|
let (report_tx, report_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
|
|
let (broadcast_tx, broadcast_rx) = mpsc::channel(IPC_CHANNEL_BUFFER);
|
|
(
|
|
Ipc {
|
|
push_tx,
|
|
queue_tx,
|
|
report_tx,
|
|
broadcast_tx: has_pubsub.then_some(broadcast_tx),
|
|
task_tx: Arc::new(Notify::new()),
|
|
task_locks: Arc::new(crate::ipc::TaskLocks::default()),
|
|
train_task_controller: Arc::new(TrainTaskController::default()),
|
|
},
|
|
IpcReceivers {
|
|
push_rx: Some(push_rx),
|
|
queue_rx: Some(queue_rx),
|
|
report_rx: Some(report_rx),
|
|
broadcast_rx: has_pubsub.then_some(broadcast_rx),
|
|
},
|
|
)
|
|
}
|