From 8d9578ee723a03464f8d5f0a85ca308cd7100844 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Mon, 31 Aug 2026 14:36:36 +0200 Subject: [PATCH] fix(hotreload): dropping configuration channels is ignored by consumers In the context of #1536 I tried to disable hot-reloading configuration and that is currently causing the main Reloader object to be dropped and closing all its watch::Senders, which in turn causes all receiver ends to constantly return errors that are ignored by all components listening on them and flooding the logs with messages of configuration being reloaded. This is fixed by using preconditions on the `tokio::select!` macro of all `watch::Receiver` consumers. If the channel is closed (when hot-reloading is disabled), the future for the branches (the call to `.changed()`) is still created, but they are not polled for completion, effectively removing the branch from the `tokio::select!`. The `.has_changed()` method is synchronous and doesn't mark the latest value in the channel as seen, which means calling it in the loop of the consumers will not lead to them missing events. The cost of calling `.has_changed()` is a single atomic load operation, since this is monitoring configuration changes and these happen spuriously, the value should rarely change and the call should be negligible. That said, configuration checks in tight loops use a local variable for caching the state of the channel. Of note, this has been broken since the original implementation, it is only noticeable now because there are some components that become noisy when the channels are dropped. --- fact/src/bpf/mod.rs | 8 ++++++-- fact/src/endpoints.rs | 8 ++++++-- fact/src/host_scanner.rs | 14 ++++++++++---- fact/src/output/grpc.rs | 4 ++-- fact/src/output/otel.rs | 10 ++++++++-- fact/src/rate_limiter.rs | 8 ++++++-- 6 files changed, 38 insertions(+), 14 deletions(-) diff --git a/fact/src/bpf/mod.rs b/fact/src/bpf/mod.rs index 815e2431..1bffb374 100644 --- a/fact/src/bpf/mod.rs +++ b/fact/src/bpf/mod.rs @@ -302,6 +302,7 @@ impl Bpf { task_set.spawn(async move { let rb = self.take_ringbuffer()?; let mut fd = AsyncFd::new(rb)?; + let mut config_is_closed = false; loop { tokio::select! { @@ -339,8 +340,11 @@ impl Bpf { } guard.clear_ready(); }, - _ = self.paths_config.changed() => { - self.load_paths().context("Failed to load paths")?; + res = self.paths_config.changed(), if !config_is_closed => { + match res { + Ok(()) => self.load_paths().context("Failed to load paths")?, + Err(_) => config_is_closed = true, + } }, _ = self.running.changed() => { if !*self.running.borrow() { diff --git a/fact/src/endpoints.rs b/fact/src/endpoints.rs index 34d95334..e1d8b6e3 100644 --- a/fact/src/endpoints.rs +++ b/fact/src/endpoints.rs @@ -74,7 +74,7 @@ impl Server { /// Wait for configuration changes or fact to stop. async fn idle(&mut self) -> anyhow::Result { tokio::select! { - _ = self.config.changed() => Ok(true), + _ = self.config.changed(), if self.config.has_changed().is_ok() => Ok(true), _ = self.running.changed() => Ok(*self.running.borrow()), } } @@ -98,7 +98,11 @@ impl Server { } }); }, - _ = self.config.changed() => return Ok(true), + res = self.config.changed(), if self.config.has_changed().is_ok() => { + if res.is_ok() { + return Ok(true); + } + } _ = self.running.changed() => return Ok(*self.running.borrow()), } } diff --git a/fact/src/host_scanner.rs b/fact/src/host_scanner.rs index ccebbc33..d0a5a15d 100644 --- a/fact/src/host_scanner.rs +++ b/fact/src/host_scanner.rs @@ -578,7 +578,7 @@ You can increase this limit with: tokio::select! { _ = interval.tick() => scan_trigger.notify_one(), _ = running.changed() => break, - _ = scan_interval.changed() => break, + _ = scan_interval.changed(), if scan_interval.has_changed().is_ok() => break, } } } @@ -611,6 +611,7 @@ You can increase this limit with: task_set.spawn(async move { info!("Starting host scanner..."); + let mut config_is_closed = false; loop { tokio::select! { @@ -708,10 +709,15 @@ You can increase this limit with: } } _ = scan_trigger.notified() => self.scan()?, - _ = self.paths.changed() => { - self.reload_paths_config()?; - self.scan()?; + res = self.paths.changed(), if !config_is_closed => { + match res { + Ok(()) => { + self.reload_paths_config()?; + self.scan()?; + } + Err(_) => config_is_closed = true, } + } } } diff --git a/fact/src/output/grpc.rs b/fact/src/output/grpc.rs index e91eb897..08ad5fe6 100644 --- a/fact/src/output/grpc.rs +++ b/fact/src/output/grpc.rs @@ -262,7 +262,7 @@ impl Client { Err(e) => warn!("gRPC stream error: {e:?}"), } } - _ = self.config.changed() => return Ok(true), + _ = self.config.changed(), if self.config.has_changed().is_ok() => return Ok(true), _ = self.running.changed() => return Ok(*self.running.borrow()), } } @@ -274,7 +274,7 @@ impl Client { async fn idle(&mut self) -> anyhow::Result { tokio::select! { - _ = self.config.changed() => Ok(true), + _ = self.config.changed(), if self.config.has_changed().is_ok() => Ok(true), _ = self.running.changed() => Ok(*self.running.borrow()), } } diff --git a/fact/src/output/otel.rs b/fact/src/output/otel.rs index 3af202df..e02dea12 100644 --- a/fact/src/output/otel.rs +++ b/fact/src/output/otel.rs @@ -77,6 +77,7 @@ impl Client { let (tx, rx) = oneshot::channel(); self.subscriber.send(tx).await?; let mut rx = rx.await?; + let mut config_is_closed = false; let res = loop { tokio::select! { @@ -109,7 +110,12 @@ impl Client { } } } - _ = self.config.changed() => break Ok(true), + res = self.config.changed(), if !config_is_closed => { + match res { + Ok(()) => break Ok(true), + Err(_) => config_is_closed = true, + } + } _ = self.running.changed() => break Ok(*self.running.borrow()), } }; @@ -124,7 +130,7 @@ impl Client { async fn idle(&mut self) -> anyhow::Result { tokio::select! { - _ = self.config.changed() => Ok(true), + _ = self.config.changed(), if self.config.has_changed().is_ok() => Ok(true), _ = self.running.changed() => Ok(*self.running.borrow()), } } diff --git a/fact/src/rate_limiter.rs b/fact/src/rate_limiter.rs index e4bd9978..f5cbffac 100644 --- a/fact/src/rate_limiter.rs +++ b/fact/src/rate_limiter.rs @@ -65,6 +65,7 @@ impl RateLimiter { pub fn start(mut self, task_set: &mut JoinSet>) { task_set.spawn(async move { debug!("Starting rate limiter..."); + let mut config_is_closed = false; loop { tokio::select! { event = self.rx.recv() => { @@ -81,8 +82,11 @@ impl RateLimiter { self.metrics.errored(); } }, - _ = self.rate_config.changed() => { - self.reload_limiter()?; + res = self.rate_config.changed(), if !config_is_closed => { + match res { + Ok(()) => self.reload_limiter()?, + Err(_) => config_is_closed = true, + } }, } }