diff --git a/crates/coordinator/Cargo.toml b/crates/coordinator/Cargo.toml index ea63a67..2a88b21 100644 --- a/crates/coordinator/Cargo.toml +++ b/crates/coordinator/Cargo.toml @@ -8,7 +8,7 @@ store = { path = "../store" } registry = { path = "../registry" } trc = { path = "../trc" } futures = { version = "0.3", optional = true } -tokio = { version = "1.53", features = ["sync", "fs", "io-util"] } +tokio = { version = "1.53", features = ["sync", "fs", "io-util", "rt", "time"] } async-nats = { version = "0.50", default-features = false, features = ["server_2_10", "server_2_11", "aws-lc-rs"], optional = true } zenoh = { version = "1.10.0", default-features = false, features = ["auth_pubkey", "transport_multilink", "transport_compression", "transport_quic", "transport_tcp", "transport_tls", "transport_udp"], optional = true } rdkafka = { version = "0.39", features = ["cmake-build"], optional = true } diff --git a/crates/coordinator/src/backend/nats/mod.rs b/crates/coordinator/src/backend/nats/mod.rs index ee01fd5..6d4d918 100644 --- a/crates/coordinator/src/backend/nats/mod.rs +++ b/crates/coordinator/src/backend/nats/mod.rs @@ -2,13 +2,22 @@ * 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 std::sync::Arc; +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; use crate::Coordinator; use async_nats::Client; use registry::schema::structs::NatsCoordinator; +use trc::ClusterEvent; pub mod pubsub; @@ -47,9 +56,116 @@ impl NatsPubSub { opts = opts.token(credentials); } + // inbuxa: connect in the background and keep trying, so a node that + // starts while NATS is down still joins the cluster once NATS is + // back, instead of running without a coordinator until restarted; + // and report the connection going and coming back + let reporter = Arc::new(Reporter::default()); + opts = opts.retry_on_initial_connect().event_callback({ + let reporter = reporter.clone(); + move |event| { + let reporter = reporter.clone(); + async move { reporter.report(event) } + } + }); + let connection_timeout = config.timeout_connection.into_inner(); + async_nats::connect_with_options(config.addresses.into_inner(), opts) .await - .map(|client| Coordinator::Nats(Arc::new(NatsPubSub { client }))) + .map(|client| { + reporter.watch_first_connection(client.clone(), connection_timeout); + Coordinator::Nats(Arc::new(NatsPubSub { client })) + }) .map_err(|err| format!("Failed to connect to Nats: {}", err)) } + + /// inbuxa: whether the client is connected to a NATS server right now. + pub fn is_connected(&self) -> bool { + matches!( + self.client.connection_state(), + async_nats::connection::State::Connected + ) + } +} + +/// inbuxa: reports the client's connection events as the server's own. +#[derive(Default)] +struct Reporter { + connected_once: AtomicBool, + // A failed attempt raises an error each time the client retries, every + // few seconds while NATS is down: report the first after each change + error_reported: AtomicBool, +} + +impl Reporter { + fn report(&self, event: async_nats::Event) { + match event { + async_nats::Event::Connected => { + self.connected_once.store(true, Ordering::Relaxed); + self.error_reported.store(false, Ordering::Relaxed); + trc::event!(Cluster(ClusterEvent::CoordinatorConnected), Type = "nats"); + } + async_nats::Event::Disconnected => { + self.error_reported.store(false, Ordering::Relaxed); + trc::event!( + Cluster(ClusterEvent::CoordinatorDisconnected), + Type = "nats", + Details = "Connection lost; reconnecting in the background", + ); + } + async_nats::Event::Closed => { + trc::event!( + Cluster(ClusterEvent::CoordinatorDisconnected), + Type = "nats", + Details = "Connection closed; no further attempts will be made", + ); + } + async_nats::Event::ClientError(async_nats::ClientError::MaxReconnects) => { + trc::event!( + Cluster(ClusterEvent::CoordinatorDisconnected), + Type = "nats", + Details = "Gave up reconnecting (maxReconnects reached)", + ); + } + async_nats::Event::ClientError(err) => { + if !self.error_reported.swap(true, Ordering::Relaxed) { + trc::event!( + Cluster(ClusterEvent::CoordinatorError), + Type = "nats", + Details = "Connection attempt failed; retrying", + Reason = err.to_string(), + ); + } + } + event => { + trc::event!( + Cluster(ClusterEvent::CoordinatorError), + Type = "nats", + Details = event.to_string(), + ); + } + } + } + + /// The first connection is made in the background, so say so when it + /// hasn't been made within the connection timeout. The client keeps + /// trying, and reports the connection when it comes. + fn watch_first_connection(self: &Arc, client: Client, timeout: Duration) { + let reporter = self.clone(); + tokio::spawn(async move { + tokio::time::sleep(timeout).await; + if !reporter.connected_once.load(Ordering::Relaxed) + && !matches!( + client.connection_state(), + async_nats::connection::State::Connected + ) + { + trc::event!( + Cluster(ClusterEvent::CoordinatorDisconnected), + Type = "nats", + Details = "Not connected at startup; retrying in the background", + ); + } + }); + } } diff --git a/crates/coordinator/src/dispatch.rs b/crates/coordinator/src/dispatch.rs index bd84b67..6f10adb 100644 --- a/crates/coordinator/src/dispatch.rs +++ b/crates/coordinator/src/dispatch.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::{Coordinator, Msg, PubSubStream}; @@ -43,6 +45,17 @@ impl Coordinator { pub fn is_none(&self) -> bool { matches!(self, Coordinator::None) } + + /// inbuxa: whether the coordinator is connected right now, for the + /// backends that track it (NATS); `None` for the others and when no + /// coordinator is configured. + pub fn is_connected(&self) -> Option { + match self { + #[cfg(feature = "nats")] + Coordinator::Nats(store) => Some(store.is_connected()), + _ => None, + } + } } impl PubSubStream { diff --git a/crates/http/src/request.rs b/crates/http/src/request.rs index 5366e6c..576b003 100644 --- a/crates/http/src/request.rs +++ b/crates/http/src/request.rs @@ -562,6 +562,27 @@ impl ParseHttp for Server { }) .into_http_response()); } + // inbuxa: the cluster coordinator's connection, for + // monitoring. It stays out of live and ready on purpose: + // a node without its coordinator still serves mail, and + // failing those would have an orchestrator restart, or + // take out of service, every node at once when the + // coordinator goes down + "cluster" => { + let coordinator = &self.core.storage.coordinator; + let (status, state) = match coordinator.is_connected() { + Some(true) => (StatusCode::OK, "connected"), + Some(false) => (StatusCode::SERVICE_UNAVAILABLE, "disconnected"), + None if coordinator.is_none() => (StatusCode::OK, "none"), + None => (StatusCode::OK, "unknown"), + }; + return Ok(http_proto::JsonResponse::with_status( + status, + serde_json::json!({ "coordinator": state }), + ) + .no_cache() + .into_http_response()); + } _ => (), } } diff --git a/crates/trc/src/event/enums.rs b/crates/trc/src/event/enums.rs index 1725ed7..dd5ee4b 100644 --- a/crates/trc/src/event/enums.rs +++ b/crates/trc/src/event/enums.rs @@ -10,8 +10,9 @@ // inbuxa: 637 to 641 are the fork's SCIM events (SCIM-54); 642 is // auth.legacy-protocol-refused (legacy-protocols LP-6); 643 is -// security.legacy-protocols-changed (LP-8) -pub const TOTAL_EVENT_COUNT: usize = 644; +// security.legacy-protocols-changed (LP-8); 644 to 646 are the cluster +// coordinator's connection events +pub const TOTAL_EVENT_COUNT: usize = 647; pub const TOTAL_METRIC_COUNT: usize = 369; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -150,6 +151,10 @@ pub enum ClusterEvent { MessageSkipped = 47, MessageInvalid = 49, NodeIdRenewed = 275, + // inbuxa: the coordinator's connection + CoordinatorConnected = 644, + CoordinatorDisconnected = 645, + CoordinatorError = 646, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] diff --git a/crates/trc/src/event/enums_impl.rs b/crates/trc/src/event/enums_impl.rs index a0864e5..d4ebe88 100644 --- a/crates/trc/src/event/enums_impl.rs +++ b/crates/trc/src/event/enums_impl.rs @@ -81,6 +81,10 @@ impl EventType { b"cluster.message-skipped" => EventType::Cluster(ClusterEvent::MessageSkipped), b"cluster.message-invalid" => EventType::Cluster(ClusterEvent::MessageInvalid), b"cluster.node-id-renewed" => EventType::Cluster(ClusterEvent::NodeIdRenewed), + // inbuxa: coordinator connection + b"cluster.coordinator-connected" => EventType::Cluster(ClusterEvent::CoordinatorConnected), + b"cluster.coordinator-disconnected" => EventType::Cluster(ClusterEvent::CoordinatorDisconnected), + b"cluster.coordinator-error" => EventType::Cluster(ClusterEvent::CoordinatorError), b"dane.authentication-success" => EventType::Dane(DaneEvent::AuthenticationSuccess), b"dane.authentication-failure" => EventType::Dane(DaneEvent::AuthenticationFailure), b"dane.no-certificates-found" => EventType::Dane(DaneEvent::NoCertificatesFound), @@ -742,6 +746,14 @@ impl EventType { EventType::Cluster(ClusterEvent::MessageSkipped) => "cluster.message-skipped", EventType::Cluster(ClusterEvent::MessageInvalid) => "cluster.message-invalid", EventType::Cluster(ClusterEvent::NodeIdRenewed) => "cluster.node-id-renewed", + // inbuxa: coordinator connection + EventType::Cluster(ClusterEvent::CoordinatorConnected) => { + "cluster.coordinator-connected" + } + EventType::Cluster(ClusterEvent::CoordinatorDisconnected) => { + "cluster.coordinator-disconnected" + } + EventType::Cluster(ClusterEvent::CoordinatorError) => "cluster.coordinator-error", EventType::Dane(DaneEvent::AuthenticationSuccess) => "dane.authentication-success", EventType::Dane(DaneEvent::AuthenticationFailure) => "dane.authentication-failure", EventType::Dane(DaneEvent::NoCertificatesFound) => "dane.no-certificates-found", @@ -1524,6 +1536,10 @@ impl EventType { EventType::Cluster(ClusterEvent::MessageSkipped) => 47, EventType::Cluster(ClusterEvent::MessageInvalid) => 49, EventType::Cluster(ClusterEvent::NodeIdRenewed) => 275, + // inbuxa: coordinator connection + EventType::Cluster(ClusterEvent::CoordinatorConnected) => 644, + EventType::Cluster(ClusterEvent::CoordinatorDisconnected) => 645, + EventType::Cluster(ClusterEvent::CoordinatorError) => 646, EventType::Dane(DaneEvent::AuthenticationSuccess) => 67, EventType::Dane(DaneEvent::AuthenticationFailure) => 66, EventType::Dane(DaneEvent::NoCertificatesFound) => 69, @@ -2176,6 +2192,10 @@ impl EventType { 47 => Some(EventType::Cluster(ClusterEvent::MessageSkipped)), 49 => Some(EventType::Cluster(ClusterEvent::MessageInvalid)), 275 => Some(EventType::Cluster(ClusterEvent::NodeIdRenewed)), + // inbuxa: coordinator connection + 644 => Some(EventType::Cluster(ClusterEvent::CoordinatorConnected)), + 645 => Some(EventType::Cluster(ClusterEvent::CoordinatorDisconnected)), + 646 => Some(EventType::Cluster(ClusterEvent::CoordinatorError)), 67 => Some(EventType::Dane(DaneEvent::AuthenticationSuccess)), 66 => Some(EventType::Dane(DaneEvent::AuthenticationFailure)), 69 => Some(EventType::Dane(DaneEvent::NoCertificatesFound)), @@ -3114,6 +3134,10 @@ impl EventType { EventType::Auth(AuthEvent::TooManyAttempts) => Level::Warn, EventType::Calendar(CalendarEvent::AlarmFailed) => Level::Warn, EventType::Cluster(ClusterEvent::SubscriberDisconnected) => Level::Warn, + // inbuxa: coordinator connection + EventType::Cluster(ClusterEvent::CoordinatorConnected) => Level::Info, + EventType::Cluster(ClusterEvent::CoordinatorDisconnected) => Level::Warn, + EventType::Cluster(ClusterEvent::CoordinatorError) => Level::Warn, EventType::Delivery(DeliveryEvent::MissingOutboundHostname) => Level::Warn, EventType::Delivery(DeliveryEvent::ConcurrencyLimitExceeded) => Level::Warn, EventType::Delivery(DeliveryEvent::RateLimitExceeded) => Level::Warn, @@ -3244,6 +3268,10 @@ impl EventType { EventType::Cluster(ClusterEvent::MessageSkipped) => "PubSub message skipped", EventType::Cluster(ClusterEvent::MessageInvalid) => "Invalid PubSub message", EventType::Cluster(ClusterEvent::NodeIdRenewed) => "Node ID renewed", + // inbuxa: coordinator connection + EventType::Cluster(ClusterEvent::CoordinatorConnected) => "Coordinator connected", + EventType::Cluster(ClusterEvent::CoordinatorDisconnected) => "Coordinator unavailable", + EventType::Cluster(ClusterEvent::CoordinatorError) => "Coordinator error", EventType::Dane(DaneEvent::AuthenticationSuccess) => "DANE authentication successful", EventType::Dane(DaneEvent::AuthenticationFailure) => "DANE authentication failed", EventType::Dane(DaneEvent::NoCertificatesFound) => "No certificates found for DANE", @@ -4322,6 +4350,10 @@ impl EventType { EventType::Cluster(ClusterEvent::MessageSkipped), EventType::Cluster(ClusterEvent::MessageInvalid), EventType::Cluster(ClusterEvent::NodeIdRenewed), + // inbuxa: coordinator connection + EventType::Cluster(ClusterEvent::CoordinatorConnected), + EventType::Cluster(ClusterEvent::CoordinatorDisconnected), + EventType::Cluster(ClusterEvent::CoordinatorError), EventType::Dane(DaneEvent::AuthenticationSuccess), EventType::Dane(DaneEvent::AuthenticationFailure), EventType::Dane(DaneEvent::NoCertificatesFound), diff --git a/resources/schema/schema.json.gz b/resources/schema/schema.json.gz index 09e6ff7..daec882 100644 Binary files a/resources/schema/schema.json.gz and b/resources/schema/schema.json.gz differ diff --git a/resources/schema/schema.json.sha256 b/resources/schema/schema.json.sha256 index 10bece9..1e11c67 100644 --- a/resources/schema/schema.json.sha256 +++ b/resources/schema/schema.json.sha256 @@ -1 +1 @@ -VbnFuwCOTBh0s2T-NuRhb2JaJr8Jl5s3LgXv4Pv2sTg \ No newline at end of file +XFI3xuKC_rH1KZyaVBF0uTIiRDXRqyYboijquiGz2eg \ No newline at end of file diff --git a/tests/src/cluster/coordinator.rs b/tests/src/cluster/coordinator.rs new file mode 100644 index 0000000..fcb25d7 --- /dev/null +++ b/tests/src/cluster/coordinator.rs @@ -0,0 +1,145 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! A node that starts while its NATS coordinator is down joins the cluster +//! once NATS comes up, without a restart, and reports the coordinator's +//! connection on `/healthz/cluster` as it goes and comes back. + +use crate::utils::server::TestServerBuilder; +use coordinator::Coordinator; +use registry::{ + schema::{ + enums::NetworkListenerProtocol, + structs::{Coordinator as CoordinatorSetting, NatsCoordinator}, + }, + types::map::Map, +}; +use serde_json::{Value, json}; +use std::time::{Duration, Instant}; +use testcontainers::{ + GenericImage, ImageExt, core::IntoContainerPort, core::WaitFor, runners::AsyncRunner, +}; + +const HTTP_PORT: u16 = 11_310; +const TOPIC: &str = "inbuxa-coordinator-test"; + +#[tokio::test(flavor = "multi_thread")] +pub async fn coordinator_reconnect_tests() { + println!("Running coordinator reconnect tests..."); + + // A port with no NATS server behind it, yet + let nats_port = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let config = NatsCoordinator { + addresses: Map::new(vec![format!("127.0.0.1:{nats_port}")]), + use_tls: false, + timeout_connection: 1_000u64.into(), + ..Default::default() + }; + + // 1. The node starts, without a build error, while NATS is down, and + // says so + let test = TestServerBuilder::new("coordinator_reconnect_tests") + .await + .with_object(CoordinatorSetting::Nats(config.clone())) + .await + .with_listener(NetworkListenerProtocol::Http, "http", HTTP_PORT, true) + .await + .build() + .await; + let coordinator = test.server.core.storage.coordinator.clone(); + assert!( + coordinator.is_enabled(), + "a coordinator, though not connected" + ); + assert_eq!(coordinator.is_connected(), Some(false)); + assert_eq!( + cluster_health().await, + (503, json!({"coordinator": "disconnected"})) + ); + + // A subscription made now, as the broadcast subscriber makes it at + // startup, has to work once NATS is up + let mut stream = coordinator.subscribe(TOPIC).await.unwrap(); + + // 2. NATS comes up: the node connects on its own + let nats = GenericImage::new("nats", "latest") + .with_wait_for(WaitFor::message_on_stderr("Server is ready")) + .with_mapped_port(nats_port, 4222.tcp()) + .start() + .await + .expect("Failed to start NATS container"); + wait_for_health(200, "connected").await; + let other_node = coordinator::backend::nats::NatsPubSub::open(config.clone()) + .await + .unwrap(); + wait_until_connected(&other_node).await; + round_trip(&other_node, &mut stream, b"after startup").await; + + // 3. NATS goes away: the node reports it; and it comes back: the node + // reconnects and the same subscription carries on + nats.stop().await.unwrap(); + wait_for_health(503, "disconnected").await; + nats.start().await.unwrap(); + wait_for_health(200, "connected").await; + wait_until_connected(&other_node).await; + round_trip(&other_node, &mut stream, b"after reconnect").await; + + drop(nats); + if test.is_reset() { + test.temp_dir.delete(); + } +} + +async fn cluster_health() -> (u16, Value) { + let response = reqwest::Client::builder() + .danger_accept_invalid_certs(true) + .timeout(Duration::from_secs(5)) + .build() + .unwrap() + .get(format!("https://127.0.0.1:{HTTP_PORT}/healthz/cluster")) + .send() + .await + .unwrap(); + let status = response.status().as_u16(); + (status, response.json().await.unwrap()) +} + +async fn wait_for_health(status: u16, state: &str) { + let started = Instant::now(); + loop { + let health = cluster_health().await; + if health == (status, json!({"coordinator": state})) { + return; + } + assert!( + started.elapsed() < Duration::from_secs(30), + "expected {status} {state}, still {health:?}" + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + +async fn wait_until_connected(coordinator: &Coordinator) { + let started = Instant::now(); + while coordinator.is_connected() != Some(true) { + assert!(started.elapsed() < Duration::from_secs(30), "not connected"); + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +/// Another node publishes; this one's subscription receives it. +async fn round_trip(from: &Coordinator, stream: &mut coordinator::PubSubStream, payload: &[u8]) { + from.publish(TOPIC, payload.to_vec()).await.unwrap(); + let message = tokio::time::timeout(Duration::from_secs(10), stream.next()) + .await + .expect("no message within 10 seconds") + .expect("subscription ended"); + assert_eq!(message.payload(), payload); +} diff --git a/tests/src/cluster/mod.rs b/tests/src/cluster/mod.rs index 95c1604..6d06c6f 100644 --- a/tests/src/cluster/mod.rs +++ b/tests/src/cluster/mod.rs @@ -2,7 +2,11 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ pub mod broadcast; +#[cfg(feature = "nats")] +pub mod coordinator; // inbuxa: coordinator reconnects pub mod stress;