diff --git a/crates/coordinator/src/backend/nats/mod.rs b/crates/coordinator/src/backend/nats/mod.rs index 6d4d918a..8964e46e 100644 --- a/crates/coordinator/src/backend/nats/mod.rs +++ b/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, } 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 { + 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, } 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); + } +} diff --git a/crates/coordinator/src/dispatch.rs b/crates/coordinator/src/dispatch.rs index 6f10adbb..a9d1ae67 100644 --- a/crates/coordinator/src/dispatch.rs +++ b/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> { + match self { + #[cfg(feature = "nats")] + Coordinator::Nats(store) => Some(store.reconnects()), + _ => None, + } + } } impl PubSubStream { diff --git a/crates/services/src/broadcast/subscriber.rs b/crates/services/src/broadcast/subscriber.rs index 20c901fb..7293663d 100644 --- a/crates/services/src/broadcast/subscriber.rs +++ b/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, 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, 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, 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, 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, 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, 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))