From 534461584551675fdb0738ff793e78f9682f852d Mon Sep 17 00:00:00 2001 From: John Coffey Date: Thu, 8 Oct 2026 14:34:42 -0700 Subject: [PATCH] Drop local caches and reload settings when the coordinator comes back Core NATS delivers at most once. While a node's connection is down the client reconnects by itself and subscribes again, but whatever the cluster published meanwhile never arrives, and the account, token, role, domain and DAV caches only ever change on those events, so they stayed stale until a restart; a missed registry change left the running settings stale the same way. The NATS backend now counts reconnections on a watch. The broadcast subscriber drops every local cache and reloads the settings, certificates and blocked addresses on each one, and after subscribing again when its stream ended. The message caches check the change log on every use and need nothing. --- crates/coordinator/src/backend/nats/mod.rs | 60 +++++++++++++++++++-- crates/coordinator/src/dispatch.rs | 10 ++++ crates/services/src/broadcast/subscriber.rs | 60 ++++++++++++++++++++- 3 files changed, 124 insertions(+), 6 deletions(-) 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)) -- 2.54.0