Coordinator: join the cluster when NATS comes up, report the connection

A node that started while NATS was down never got a coordinator. The
connect failed at boot, bootstrap recorded a build error and the node ran
with Coordinator::None until restarted. It had no broadcast subscriber
or publisher, so cross-node push and cache invalidation to it stayed
broken, and its healthcheck said nothing about it. Losing NATS after
startup was silent too.

- The NATS client now connects in the background
  (retry_on_initial_connect): startup never waits on NATS or fails over
  it, the node gets its coordinator, subscriber and publisher at once,
  and the client keeps trying (async-nats's backoff, at most 4 s apart)
  until NATS answers. Subscriptions made meanwhile start delivering when
  it does. A configured maxReconnects still ends the attempts.
- Three new events report the connection: cluster.coordinator-connected
  (info), cluster.coordinator-disconnected (warn: lost, closed, gave up,
  or not connected within the connection timeout at startup) and
  cluster.coordinator-error (warn: a failed attempt, reported once per
  outage rather than every retry, and server errors, slow consumers and
  lame duck mode). They are in the packaged schema, ids 644 to 646.
- GET /healthz/cluster reports the coordinator: 200
  {"coordinator":"connected"}, 503 {"coordinator":"disconnected"}, or
  200 with "none" (no coordinator) or "unknown" (a backend that doesn't
  track its connection). /healthz/live and /healthz/ready are unchanged
  on purpose: a node without its coordinator still serves mail, and
  failing those would have orchestrators restart, or pull out of
  service, every node at once whenever NATS is down.

Only NATS connects lazily; the other coordinator backends still fail at
boot as before.

cluster::coordinator::coordinator_reconnect_tests starts a node against a
NATS port with nothing behind it, checks it boots with a coordinator and
reports it disconnected, subscribes, then starts NATS on that port: the
node connects on its own and the subscription receives a message from a
second client. Stopping and restarting NATS shows disconnected, then
connected, and the same subscription keeps working.

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 }

crates/coordinator/src/backend/nats/mod.rs

@@ -2,13 +2,22 @@
  * 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 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<Self>, 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",
+                );
+            }
+        });
+    }
 }

crates/coordinator/src/dispatch.rs

@@ -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::{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<bool> {
+        match self {
+            #[cfg(feature = "nats")]
+            Coordinator::Nats(store) => Some(store.is_connected()),
+            _ => None,
+        }
+    }
 }
 
 impl PubSubStream {