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

@@ -17,6 +17,7 @@ use std::{
 use crate::Coordinator;
 use async_nats::Client;
 use registry::schema::structs::NatsCoordinator;
+use tokio::sync::watch;
 use trc::ClusterEvent;
 
 pub mod pubsub;
@@ -24,6 +25,7 @@ pub mod pubsub;
 #[derive(Debug)]
 pub struct NatsPubSub {
     client: Client,
+    reporter: Arc<Reporter>,
 }
 
 impl NatsPubSub {
@@ -60,7 +62,7 @@ impl NatsPubSub {
         // 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());
+        let reporter = Arc::new(Reporter::new());
         opts = opts.retry_on_initial_connect().event_callback({
             let reporter = reporter.clone();
             move |event| {
@@ -74,11 +76,18 @@ impl NatsPubSub {
             .await
             .map(|client| {
                 reporter.watch_first_connection(client.clone(), connection_timeout);
-                Coordinator::Nats(Arc::new(NatsPubSub { client }))
+                Coordinator::Nats(Arc::new(NatsPubSub { client, reporter }))
             })
             .map_err(|err| format!("Failed to connect to Nats: {}", err))
     }
 
+    /// inbuxa: counts the client's reconnections. Core NATS delivers at
+    /// most once, so whatever the cluster published while this node was
+    /// away is gone; a watcher drops what those events would have changed.
+    pub fn reconnects(&self) -> watch::Receiver<u64> {
+        self.reporter.reconnects.subscribe()
+    }
+
     /// inbuxa: whether the client is connected to a NATS server right now.
     pub fn is_connected(&self) -> bool {
         matches!(
@@ -89,21 +98,40 @@ impl NatsPubSub {
 }
 
 /// inbuxa: reports the client's connection events as the server's own.
-#[derive(Default)]
+#[derive(Debug)]
 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,
+    /// How many times the connection came back after being lost.
+    reconnects: watch::Sender<u64>,
 }
 
 impl Reporter {
+    fn new() -> Self {
+        Self {
+            connected_once: AtomicBool::new(false),
+            error_reported: AtomicBool::new(false),
+            reconnects: watch::Sender::new(0),
+        }
+    }
+
     fn report(&self, event: async_nats::Event) {
         match event {
             async_nats::Event::Connected => {
-                self.connected_once.store(true, Ordering::Relaxed);
+                let reconnected = self.connected_once.swap(true, Ordering::Relaxed);
                 self.error_reported.store(false, Ordering::Relaxed);
-                trc::event!(Cluster(ClusterEvent::CoordinatorConnected), Type = "nats");
+                if reconnected {
+                    self.reconnects.send_modify(|count| *count += 1);
+                    trc::event!(
+                        Cluster(ClusterEvent::CoordinatorConnected),
+                        Type = "nats",
+                        Details = "Reconnected",
+                    );
+                } else {
+                    trc::event!(Cluster(ClusterEvent::CoordinatorConnected), Type = "nats");
+                }
             }
             async_nats::Event::Disconnected => {
                 self.error_reported.store(false, Ordering::Relaxed);
@@ -169,3 +197,25 @@ impl Reporter {
         });
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn only_a_connection_that_comes_back_counts() {
+        let reporter = Reporter::new();
+        let watcher = reporter.reconnects.subscribe();
+        reporter.report(async_nats::Event::Connected);
+        assert_eq!(
+            *watcher.borrow(),
+            0,
+            "the first connection is not a reconnect"
+        );
+        reporter.report(async_nats::Event::Disconnected);
+        reporter.report(async_nats::Event::Connected);
+        assert_eq!(*watcher.borrow(), 1);
+        reporter.report(async_nats::Event::Connected);
+        assert_eq!(*watcher.borrow(), 2);
+    }
+}

crates/coordinator/src/dispatch.rs

@@ -56,6 +56,16 @@ impl Coordinator {
             _ => None,
         }
     }
+
+    /// inbuxa: a watch on the number of reconnections, for backends that
+    /// report them (NATS); `None` for the others.
+    pub fn reconnects(&self) -> Option<tokio::sync::watch::Receiver<u64>> {
+        match self {
+            #[cfg(feature = "nats")]
+            Coordinator::Nats(store) => Some(store.reconnects()),
+            _ => None,
+        }
+    }
 }
 
 impl PubSubStream {

crates/services/src/broadcast/subscriber.rs

@@ -11,7 +11,7 @@ use common::{
     BuildServer, Inner,
     ipc::{BroadcastEvent, PushEvent, PushNotification, QueueEvent, RegistryChange},
 };
-use registry::types::EnumImpl;
+use registry::{schema::prelude::ObjectType, types::EnumImpl};
 use std::{sync::Arc, time::Duration};
 use tokio::sync::watch;
 use trc::{ClusterEvent, ServerEvent};
@@ -27,6 +27,9 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
 
     tokio::spawn(async move {
         let mut retry_count: u32 = 0;
+        // inbuxa: whether events may have been published while this node
+        // had no subscription
+        let mut lost_events = false;
 
         trc::event!(Cluster(ClusterEvent::SubscriberStart));
 
@@ -43,6 +46,9 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
             let mut stream = match coordinator.subscribe(BROADCAST_TOPIC).await {
                 Ok(stream) => {
                     retry_count = 0;
+                    if std::mem::take(&mut lost_events) {
+                        recover_missed_events(&inner, "Subscribed again").await;
+                    }
                     stream
                 }
                 Err(err) => {
@@ -69,8 +75,28 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
                 }
             };
 
+            // inbuxa: the client subscribes again by itself after a
+            // reconnection, but core NATS delivers at most once, so what the
+            // cluster published meanwhile never arrives
+            let mut reconnects = coordinator.reconnects();
+            if let Some(reconnects) = &mut reconnects {
+                reconnects.mark_unchanged();
+            }
+
             loop {
                 tokio::select! {
+                    changed = async {
+                        match &mut reconnects {
+                            Some(reconnects) => reconnects.changed().await.is_ok(),
+                            None => std::future::pending().await,
+                        }
+                    } => {
+                        if changed {
+                            recover_missed_events(&inner, "Reconnected to the coordinator").await;
+                        } else {
+                            reconnects = None;
+                        }
+                    },
                     message = stream.next() => {
                         match message {
                             Some(message) => {
@@ -234,6 +260,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
                                 trc::event!(
                                     Cluster(ClusterEvent::SubscriberDisconnected),
                                 );
+                                lost_events = true;
                                 break;
                             }
                         }
@@ -249,6 +276,37 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
     });
 }
 
+/// inbuxa: after a gap in which cluster events may have been missed, drops
+/// every local cache that only those events keep fresh and rebuilds the
+/// settings, certificates and blocked addresses from the registry, as the
+/// missed events would have. The message caches check the change log on
+/// each use and need nothing.
+async fn recover_missed_events(inner: &Arc<Inner>, why: &str) {
+    let server = inner.build_server();
+    server.invalidate_all_local_caches();
+    for object in [
+        ObjectType::ClusterRole,
+        ObjectType::Certificate,
+        ObjectType::BlockedIp,
+    ] {
+        match Box::pin(server.reload_registry(RegistryChange::Reload(object))).await {
+            Ok(result) => result.log(),
+            Err(err) => {
+                trc::error!(
+                    err.details("Failed to reload settings after the coordinator came back")
+                        .caused_by(trc::location!())
+                );
+            }
+        }
+    }
+    trc::event!(
+        Cluster(ClusterEvent::CoordinatorConnected),
+        Details = format!(
+            "{why}: events may have been missed; local caches dropped and settings reloaded"
+        ),
+    );
+}
+
 /// Delay before the next subscribe attempt: 1 s, 2 s, 4 s ... capped at 64 s.
 fn subscribe_retry_delay(retry_count: u32) -> Duration {
     Duration::from_secs(1u64 << retry_count.min(6))