From 21da4e007450e0a0750ae712ac01f0196d428514 Mon Sep 17 00:00:00 2001 From: riccardo-negri <67798955+riccardo-negri@users.noreply.github.com> Date: Mon, 3 Aug 2026 10:24:08 +0200 Subject: [PATCH 1/4] feat(netconf-proto): add global YANG module cache Add `YangModuleCache`, a thread-safe cache of raw module texts shared by all SSH sessions so `get_yang_module` skips redundant `get-schema` RPCs for a given `(name, revision)`. Only versioned modules are cached; values are stored as `Arc` for cheap hits. Exposes hit/miss/size counters via OTel, and renames `YangSchemaFormat` -> `GetSchemaFormat` per RFC 6022. --- crates/collector/src/lib.rs | 11 +- crates/netconf-proto/src/client.rs | 74 ++++- crates/netconf-proto/src/lib.rs | 2 + crates/netconf-proto/src/protocol.rs | 26 +- crates/netconf-proto/src/yang_module_cache.rs | 262 ++++++++++++++++++ crates/yang-push/src/cache/actor.rs | 52 +++- crates/yang-push/src/cache/fetcher.rs | 10 +- 7 files changed, 406 insertions(+), 31 deletions(-) create mode 100644 crates/netconf-proto/src/yang_module_cache.rs diff --git a/crates/collector/src/lib.rs b/crates/collector/src/lib.rs index 50412619..6db85a1c 100644 --- a/crates/collector/src/lib.rs +++ b/crates/collector/src/lib.rs @@ -34,6 +34,7 @@ use netcalyx_bmp_service::supervisor::BmpSupervisorHandle; use netcalyx_flow_pkt::FlowInfo; use netcalyx_flow_service::FlowRequest; use netcalyx_flow_service::flow_supervisor::FlowCollectorsSupervisorActorHandle; +use netcalyx_netconf_proto::yang_module_cache::YangModuleCache; use netcalyx_udp_notif_pkt::raw::MediaType; use netcalyx_udp_notif_service::UdpNotifRequest; use netcalyx_udp_notif_service::supervisor::UdpNotifSupervisorHandle; @@ -556,12 +557,14 @@ pub async fn init_udp_notif_collection( // Only one schema cache is needed for all publishers let cache_location: PathBuf = udp_notif_config.cache_location.into(); - let netconf_fetcher = netconf_fetcher(&udp_notif_config.netconf)?; + let global_module_cache = YangModuleCache::new(); + let netconf_fetcher = netconf_fetcher(&udp_notif_config.netconf, global_module_cache.clone())?; let (_schema_join, schema_handle) = CacheActorHandle::new( 10000, either::Right(cache_location), netconf_fetcher, Duration::from_mins(5), + global_module_cache, either::Left(meter.clone()), )?; @@ -992,7 +995,10 @@ fn serialize_bmp( Ok((Some(key), value)) } -fn netconf_fetcher(config: &NetconfConfig) -> Result { +fn netconf_fetcher( + config: &NetconfConfig, + global_module_cache: YangModuleCache, +) -> Result { let user = &config.username; let private_key_path: PathBuf = (&config.private_key_path).into(); @@ -1021,6 +1027,7 @@ fn netconf_fetcher(config: &NetconfConfig) -> Result { /// to [`crate::codec::DEFAULT_MAX_MESSAGE_SIZE`], override with /// [`NetconfSshConnectConfig::with_max_message_size`]. max_message_size: usize, + module_cache: YangModuleCache, } impl NetconfSshConnectConfig { - pub const fn new( + pub fn new( auth: SshAuth, peer_address: SocketAddr, local_address: Option, @@ -174,6 +176,7 @@ impl NetconfSshConnectConfig { handler, config, max_message_size: crate::codec::DEFAULT_MAX_MESSAGE_SIZE, + module_cache: YangModuleCache::new(), } } @@ -184,6 +187,11 @@ impl NetconfSshConnectConfig { self } + pub fn with_module_cache(mut self, module_cache: YangModuleCache) -> Self { + self.module_cache = module_cache; + self + } + pub const fn auth(&self) -> &SshAuth { &self.auth } @@ -307,6 +315,7 @@ where stream, config.announce_caps, config.max_message_size, + config.module_cache, ) .await } @@ -334,6 +343,11 @@ pub struct NetConfSshClient { /// making multiple requests to the device to get the filters when /// processing multiple subscriptions yang_push_filters: Option, + + /// Global YANG module cache shared across all sessions that use the same + /// [`YangModuleCache`] instance. Populated transparently by + /// [`get_yang_module`](Self::get_yang_module); callers see no difference. + module_cache: YangModuleCache, } impl NetConfSshClient { @@ -362,6 +376,10 @@ impl NetConfSshClient { pub fn yang_library(&self) -> Option> { self.yang_library.as_ref().map(Arc::clone) } + + pub fn module_cache(&self) -> &YangModuleCache { + &self.module_cache + } } impl NetConfSshClient { @@ -419,6 +437,7 @@ impl NetConfSshClient { stream: T, announce_caps: HashSet, max_message_size: usize, + module_cache: YangModuleCache, ) -> Result { let framed = Framed::new(stream, SshCodec::with_max_message_size(max_message_size)); let (framed, session_id, peer_caps) = Self::exchange_hello(framed, announce_caps).await?; @@ -431,6 +450,7 @@ impl NetConfSshClient { next_message_id, yang_library: None, yang_push_filters: None, + module_cache, }) } @@ -503,20 +523,37 @@ impl NetConfSshClient { Ok(()) } - /// Get YANG schema from the device - pub async fn get_schema( + /// Fetch a YANG module from the device via the NETCONF `get-schema` RPC, + /// consulting the shared module cache first. + /// + /// Caching requires a `version`: the shared cache is keyed by + /// `(name, revision)`, so only modules that carry a revision are cached. + /// When `version` is `Some`, a cache hit skips the RPC entirely and a miss + /// populates the cache for all future calls. When `version` is `None` the + /// module is fetched straight from the device on every call, bypassing the + /// cache. + pub async fn get_yang_module( &mut self, name: &str, version: Option<&str>, - ) -> Result, NetConfSshClientError> { + ) -> Result, NetConfSshClientError> { + if let Some(version) = version + && let Some(cached) = self.module_cache.get(name, version) + { + trace!( + "[{}] yang module cache hit for `{name}` revision {version}", + self.peer + ); + return Ok(cached); + } debug!( - "[{}] Getting a YANG schema with name `{name}` and version {version:?}", + "[{}] Getting a YANG module with name `{name}` and version {version:?}", self.peer ); let rpc = RpcOperation::WellKnown(WellKnownOperation::GetSchema { identifier: name.into(), version: version.map(Into::into), - format: Some(YangSchemaFormat::Yang), + format: Some(GetSchemaFormat::Yang), }); let message_id = self.rpc(rpc).await?; let rpc_reply = self.rpc_reply().await?; @@ -532,7 +569,13 @@ impl NetConfSshClient { if let RpcResponse::WellKnown(WellKnownRpcResponse::YangSchema { schema }) = rpc_response { - return Ok(schema); + let arc: Arc = Arc::from(schema.as_ref()); + // Only versioned modules are cached; the cache is keyed by + // `(name, revision)`. + if let Some(version) = version { + self.module_cache.insert(name, version, Arc::clone(&arc)); + } + return Ok(arc); } else { unreachable!() } @@ -637,7 +680,9 @@ impl NetConfSshClient { visited.insert(module.name().to_string()); // Fetch the YANG schema - let schema = self.get_schema(module.name(), module.revision()).await?; + let schema = self + .get_yang_module(module.name(), module.revision()) + .await?; // Parse dependencies from schema let deps = extract_yang_dependencies(&schema).map_err(|error| { NetConfSshClientError::YangSchemaParsingError { @@ -706,17 +751,22 @@ impl NetConfSshClient { match module { ModuleType::Full(module) => { builder - .add_module(module, schema, checker) + .add_module(module, Box::from(schema.as_ref()), checker) .map_err(NetConfSshClientError::DependencyError)?; } ModuleType::FullSubmodule(module_name, submodule) => { builder - .add_submodule_for_module(module_name.as_ref(), submodule, schema, checker) + .add_submodule_for_module( + module_name.as_ref(), + submodule, + Box::from(schema.as_ref()), + checker, + ) .map_err(NetConfSshClientError::DependencyError)?; } ModuleType::ImportOnly(module) => { builder - .add_import_only_module(module, schema, checker) + .add_import_only_module(module, Box::from(schema.as_ref()), checker) .map_err(NetConfSshClientError::DependencyError)?; } ModuleType::ImportOnlySubmodule(module_name, submodule) => { @@ -724,7 +774,7 @@ impl NetConfSshClient { .add_submodule_for_import_only_module( module_name.as_ref(), submodule, - schema, + Box::from(schema.as_ref()), checker, ) .map_err(NetConfSshClientError::DependencyError)?; diff --git a/crates/netconf-proto/src/lib.rs b/crates/netconf-proto/src/lib.rs index de62f629..9b4e34fe 100644 --- a/crates/netconf-proto/src/lib.rs +++ b/crates/netconf-proto/src/lib.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -30,6 +31,7 @@ pub mod client; pub mod codec; pub mod protocol; pub mod xml_utils; +pub mod yang_module_cache; pub mod yang_push; pub mod yanglib; pub mod yangparser; diff --git a/crates/netconf-proto/src/protocol.rs b/crates/netconf-proto/src/protocol.rs index 641a19d6..e812b67e 100644 --- a/crates/netconf-proto/src/protocol.rs +++ b/crates/netconf-proto/src/protocol.rs @@ -324,7 +324,7 @@ impl XmlSerialize for Rpc { } #[derive(Eq, PartialEq, Debug, Copy, Clone, Serialize, Deserialize, strum_macros::Display)] -pub enum YangSchemaFormat { +pub enum GetSchemaFormat { #[strum(serialize = "xsd")] Xsd, @@ -341,17 +341,17 @@ pub enum YangSchemaFormat { Rnc, } -impl<'a> XmlDeserialize<'a, YangSchemaFormat> for YangSchemaFormat { +impl<'a> XmlDeserialize<'a, GetSchemaFormat> for GetSchemaFormat { fn xml_deserialize(parser: &mut XmlParser<'a, impl io::BufRead>) -> Result { parser.skip_text()?; parser.open(Some(NETCONF_MONITORING_NS), "format")?; let value_str = parser.tag_string()?; let value = match value_str.as_ref().trim() { - "xsd" => YangSchemaFormat::Xsd, - "yang" => YangSchemaFormat::Yang, - "yin" => YangSchemaFormat::Yin, - "rng" => YangSchemaFormat::Rng, - "rnc" => YangSchemaFormat::Rnc, + "xsd" => GetSchemaFormat::Xsd, + "yang" => GetSchemaFormat::Yang, + "yin" => GetSchemaFormat::Yin, + "rng" => GetSchemaFormat::Rng, + "rnc" => GetSchemaFormat::Rnc, _ => { return Err(ParsingError::InvalidValue(format!( "unknown YANG schema format `{value_str}`" @@ -363,7 +363,7 @@ impl<'a> XmlDeserialize<'a, YangSchemaFormat> for YangSchemaFormat { } } -impl XmlSerialize for YangSchemaFormat { +impl XmlSerialize for GetSchemaFormat { fn xml_serialize( &self, writer: &mut XmlWriter, @@ -842,7 +842,7 @@ pub enum WellKnownOperation { /// The data modeling language of the schema. If this parameter is not /// present, and more than one formats of the schema exists on the /// server, a 'data-not-unique' error is returned, as described above. - format: Option, + format: Option, }, } @@ -983,7 +983,7 @@ impl WellKnownOperation { None }; - let format = match YangSchemaFormat::xml_deserialize(parser) { + let format = match GetSchemaFormat::xml_deserialize(parser) { Ok(format) => Some(format), Err(ParsingError::WrongToken { expecting, .. }) if expecting == "" => None, Err(err) => return Err(err), @@ -1002,7 +1002,7 @@ impl WellKnownOperation { writer: &mut XmlWriter, identifier: &str, version: &Option>, - format: &Option, + format: &Option, ) -> Result<(), quick_xml::Error> { let mut ns_added = false; if writer.get_namespace_prefix(NETCONF_MONITORING_NS).is_none() { @@ -3148,7 +3148,7 @@ mod tests { let get_schema = WellKnownOperation::GetSchema { identifier: "foo".into(), version: Some("1.0".into()), - format: Some(YangSchemaFormat::Yang), + format: Some(GetSchemaFormat::Yang), }; test_xml_value(get_schema_str, get_schema)?; Ok(()) @@ -3168,7 +3168,7 @@ mod tests { RpcOperation::WellKnown(WellKnownOperation::GetSchema { identifier: "foo".into(), version: Some("1.0".into()), - format: Some(YangSchemaFormat::Yang), + format: Some(GetSchemaFormat::Yang), }), ); test_xml_value(get_schema_str, get_schema)?; diff --git a/crates/netconf-proto/src/yang_module_cache.rs b/crates/netconf-proto/src/yang_module_cache.rs new file mode 100644 index 00000000..8796e8b7 --- /dev/null +++ b/crates/netconf-proto/src/yang_module_cache.rs @@ -0,0 +1,262 @@ +// Copyright (C) 2026-present The NetCalyx Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +// implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Global YANG module text cache. +//! +//! The YANG specification guarantees that a `(module_name, revision)` pair +//! identifies stable content: any published change to a module MUST add a new +//! revision date ([RFC 7950], §11). A single instance of +//! [`YangModuleCache`] can therefore be shared across all routers and all SSH +//! sessions: once a module is fetched from any device, subsequent calls to +//! [`NetConfSshClient::get_yang_module`](crate::client::NetConfSshClient::get_yang_module) +//! skip the NETCONF `get-schema` RPC entirely. +//! +//! Note: the NETCONF RPC is still called `get-schema` (per [RFC 6022]), but +//! what it returns — and what we cache — is a YANG **module** text, not a +//! schema. +//! +//! ## Metrics +//! +//! [`YangModuleCache`] exposes three plain counters via [`YangModuleCacheStats`]: +//! +//! | field | meaning | +//! |-------|---------| +//! | [`YangModuleCacheStats::hits`] | `get-schema` RPC avoided (module already cached) | +//! | [`YangModuleCacheStats::misses`] | `get-schema` RPC issued (module not yet cached) | +//! | [`YangModuleCacheStats::size`] | number of distinct modules currently cached | +//! +//! These are `AtomicU64` so they can be read from any thread without holding +//! the cache lock. Higher-level crates that own an OTel meter can poll them +//! and record gauges / counters as needed. +//! +//! ## References +//! +//! - [RFC 6022]: YANG Module for NETCONF Monitoring — defines the `get-schema` +//! operation used to fetch module texts. +//! - [RFC 7950]: The YANG 1.1 Data Modeling Language — §11 "Updating a Module" +//! (any published change MUST add a new revision date). +//! +//! [RFC 6022]: https://www.rfc-editor.org/rfc/rfc6022 +//! [RFC 7950]: https://www.rfc-editor.org/rfc/rfc7950 + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, RwLock}; + +type ModuleCacheInner = Arc>>>; + +/// Metrics counters exposed by [`YangModuleCache`]. +#[derive(Debug, Default)] +pub struct YangModuleCacheStats { + /// Number of `get-schema` RPCs avoided because the module was already + /// cached. + pub hits: AtomicU64, + /// Number of `get-schema` RPCs issued because the module was not yet + /// cached. + pub misses: AtomicU64, + /// Current number of distinct `(name, revision)` entries in the cache. + pub size: AtomicU64, +} + +/// A thread-safe, globally-shared cache of raw YANG module texts. +/// +/// Keyed by `(module_name, revision)`. The value is the raw module text as +/// returned by a NETCONF `get-schema` RPC, stored as `Arc` so that cache +/// hits — and the value handed back by +/// [`NetConfSshClient::get_yang_module`](crate::client::NetConfSshClient::get_yang_module) +/// — are cheap pointer clones rather than full string copies. (Feeding a +/// module into the `ModuleSetBuilder` still costs one copy, because the builder +/// takes an owned `Box`.) +/// +/// Clone is cheap — clones share the same backing store and stats. +#[derive(Debug, Clone, Default)] +pub struct YangModuleCache { + inner: ModuleCacheInner, + stats: Arc, +} + +impl YangModuleCache { + pub fn new() -> Self { + Self::default() + } + + pub fn stats(&self) -> &Arc { + &self.stats + } + + /// Return the cached module text for `(name, revision)`, or `None` on miss. + /// Increments the appropriate stats counter. + pub fn get(&self, name: &str, revision: &str) -> Option> { + let key = Self::make_key(name, revision); + let result = self + .inner + .read() + .expect("yang module cache lock poisoned") + .get(&key) + .cloned(); + if result.is_some() { + self.stats.hits.fetch_add(1, Ordering::Relaxed); + } else { + self.stats.misses.fetch_add(1, Ordering::Relaxed); + } + result + } + + /// Insert a module text. First writer wins: if `(name, revision)` is + /// already present the call is a no-op. This is safe because identical + /// `(name, revision)` always has identical content per the YANG spec. + pub fn insert(&self, name: &str, revision: &str, text: Arc) { + let key = Self::make_key(name, revision); + let mut map = self.inner.write().expect("yang module cache lock poisoned"); + let prev_len = map.len(); + map.entry(key).or_insert(text); + if map.len() > prev_len { + self.stats.size.fetch_add(1, Ordering::Relaxed); + } + } + + /// Number of entries currently in the cache. + pub fn len(&self) -> usize { + self.inner + .read() + .expect("yang module cache lock poisoned") + .len() + } + + pub fn is_empty(&self) -> bool { + self.inner + .read() + .expect("yang module cache lock poisoned") + .is_empty() + } + + fn make_key(name: &str, revision: &str) -> String { + format!("{name}@{revision}") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_new_is_empty() { + let c = YangModuleCache::new(); + assert!(c.is_empty()); + assert_eq!(c.len(), 0); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 0); + } + + #[test] + fn test_miss_increments_miss_counter() { + let c = YangModuleCache::new(); + assert!(c.get("ietf-interfaces", "2018-02-20").is_none()); + assert_eq!(c.stats().misses.load(Ordering::Relaxed), 1); + assert_eq!(c.stats().hits.load(Ordering::Relaxed), 0); + } + + #[test] + fn test_insert_and_hit_increments_hit_counter() { + let c = YangModuleCache::new(); + c.insert( + "ietf-interfaces", + "2018-02-20", + Arc::from("module ietf-interfaces { }"), + ); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 1); + + let result = c.get("ietf-interfaces", "2018-02-20"); + assert_eq!(result.as_deref(), Some("module ietf-interfaces { }")); + assert_eq!(c.stats().hits.load(Ordering::Relaxed), 1); + assert_eq!(c.stats().misses.load(Ordering::Relaxed), 0); + } + + #[test] + fn test_first_writer_wins() { + let c = YangModuleCache::new(); + c.insert("mod", "2024-01-01", Arc::from("first")); + c.insert("mod", "2024-01-01", Arc::from("second")); + assert_eq!(c.get("mod", "2024-01-01").as_deref(), Some("first")); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 1); + } + + #[test] + fn test_different_revisions_are_distinct_keys() { + let c = YangModuleCache::new(); + c.insert("mod", "2023-01-01", Arc::from("old")); + c.insert("mod", "2024-01-01", Arc::from("new")); + assert_eq!(c.get("mod", "2023-01-01").as_deref(), Some("old")); + assert_eq!(c.get("mod", "2024-01-01").as_deref(), Some("new")); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 2); + } + + #[test] + fn test_clone_shares_state() { + let a = YangModuleCache::new(); + let b = a.clone(); + a.insert("mod", "2024-01-01", Arc::from("value")); + assert_eq!(b.get("mod", "2024-01-01").as_deref(), Some("value")); + // hit recorded on `b` is visible via `a.stats` (same Arc) + assert_eq!(a.stats().hits.load(Ordering::Relaxed), 1); + } + + #[test] + fn test_concurrent_insert_and_get() { + use std::thread; + + let cache = YangModuleCache::new(); + let n_threads = 8; + let n_modules = 20; + + let handles: Vec<_> = (0..n_threads) + .map(|t| { + let c = cache.clone(); + thread::spawn(move || { + for i in 0..n_modules { + let name = format!("mod-{i}"); + let rev = format!("2024-{i:02}-01"); + let text = Arc::from(format!("text-{i}").as_str()); + c.insert(&name, &rev, text); + assert!(c.get(&name, &rev).is_some()); + let _ = c.len(); + } + for i in 0..n_modules { + let name = format!("mod-{i}"); + let rev = format!("2024-{i:02}-01"); + c.insert( + &name, + &rev, + Arc::from(format!("other-text-{t}-{i}").as_str()), + ); + } + }) + }) + .collect(); + + for h in handles { + h.join().expect("thread panicked"); + } + + // All modules must be present and have the first-writer value. + assert_eq!(cache.len(), n_modules); + for i in 0..n_modules { + let name = format!("mod-{i}"); + let rev = format!("2024-{i:02}-01"); + let expected = format!("text-{i}"); + assert_eq!(cache.get(&name, &rev).as_deref(), Some(expected.as_str())); + } + assert_eq!(cache.stats().size.load(Ordering::Relaxed), n_modules as u64); + } +} diff --git a/crates/yang-push/src/cache/actor.rs b/crates/yang-push/src/cache/actor.rs index 29b0ce66..45482bad 100644 --- a/crates/yang-push/src/cache/actor.rs +++ b/crates/yang-push/src/cache/actor.rs @@ -223,11 +223,13 @@ use crate::{ }; use futures_util::StreamExt; use futures_util::stream::FuturesUnordered; +use netcalyx_netconf_proto::yang_module_cache::YangModuleCache; use netcalyx_netconf_proto::yang_push::types::SubscriptionId; use netcalyx_udp_notif_service::SessionInfo; use rustc_hash::FxHashMap; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::Duration; use tokio::sync::{mpsc, oneshot}; use tokio::task::{JoinError, JoinHandle}; @@ -245,10 +247,15 @@ pub struct CachingStats { pub device_fetch_queue: opentelemetry::metrics::Gauge, pub device_fetch_succeeded: opentelemetry::metrics::Counter, pub device_fetch_failed: opentelemetry::metrics::Counter, + // Observable instruments that read from YangModuleCache atomics. + // Held here to keep the OTel callbacks registered for the lifetime of the actor. + _yang_module_cache_hits: opentelemetry::metrics::ObservableCounter, + _yang_module_cache_misses: opentelemetry::metrics::ObservableCounter, + _yang_module_cache_size: opentelemetry::metrics::ObservableGauge, } impl CachingStats { - pub fn new(meter: opentelemetry::metrics::Meter) -> Self { + pub fn new(meter: opentelemetry::metrics::Meter, module_cache: &YangModuleCache) -> Self { let requests_received = meter .u64_counter("netcalyx.collector.yang_push.caching.requests.received") .with_description("Number of requests received by the YANG library cache actor") @@ -283,6 +290,41 @@ impl CachingStats { .u64_counter("netcalyx.collector.yang_push.caching.device.fetch.response.failed") .with_description("Number of device fetch requests initiated by the YANG library cache actor and failed") .build(); + + let stats = module_cache.stats().clone(); + let _yang_module_cache_hits = { + let s = Arc::clone(&stats); + meter + .u64_observable_counter("netcalyx.collector.yang_push.caching.yang_module_cache.hits") + .with_description("Number of get-schema RPCs avoided because the YANG module was already in the shared cache") + .with_callback(move |counter| { + counter.observe(s.hits.load(Ordering::Relaxed), &[]); + }) + .build() + }; + let _yang_module_cache_misses = { + let s = Arc::clone(&stats); + meter + .u64_observable_counter("netcalyx.collector.yang_push.caching.yang_module_cache.misses") + .with_description("Number of get-schema RPCs issued because the YANG module was not yet in the shared cache") + .with_callback(move |counter| { + counter.observe(s.misses.load(Ordering::Relaxed), &[]); + }) + .build() + }; + let _yang_module_cache_size = { + let s = Arc::clone(&stats); + meter + .u64_observable_gauge("netcalyx.collector.yang_push.caching.yang_module_cache.size") + .with_description( + "Number of distinct YANG modules currently held in the shared module cache", + ) + .with_callback(move |gauge| { + gauge.observe(s.size.load(Ordering::Relaxed), &[]); + }) + .build() + }; + Self { requests_received, pending_cache_requests, @@ -292,6 +334,9 @@ impl CachingStats { device_fetch_queue, device_fetch_succeeded, device_fetch_failed, + _yang_module_cache_hits, + _yang_module_cache_misses, + _yang_module_cache_size, } } } @@ -1218,6 +1263,7 @@ impl CacheActorHandle { schema_cache: either::Either, fetcher: F, fetcher_timeout: Duration, + global_module_cache: YangModuleCache, stats: either::Either, ) -> Result<(JoinHandle>, Self), CacheActorHandleError> { @@ -1228,7 +1274,7 @@ impl CacheActorHandle { either::Either::Right(root_path) => YangLibraryCache::from_disk(root_path)?, }; let stats = match stats { - either::Either::Left(meter) => CachingStats::new(meter), + either::Either::Left(meter) => CachingStats::new(meter, &global_module_cache), either::Either::Right(stats) => stats, }; @@ -1369,6 +1415,7 @@ pub(crate) mod tests { either::Right(cache_dir.path().to_path_buf()), fetcher, Duration::from_secs(1), + YangModuleCache::new(), either::Either::Left(opentelemetry::global::meter("test-meter")), ) .expect("Failed to create cache actor"); @@ -1411,6 +1458,7 @@ pub(crate) mod tests { either::Right(cache_dir.path().to_path_buf()), fetcher, Duration::from_secs(1), + YangModuleCache::new(), either::Either::Left(opentelemetry::global::meter("test-meter")), ) .expect("Failed to create cache actor"); diff --git a/crates/yang-push/src/cache/fetcher.rs b/crates/yang-push/src/cache/fetcher.rs index 6c30570e..7dec34c4 100644 --- a/crates/yang-push/src/cache/fetcher.rs +++ b/crates/yang-push/src/cache/fetcher.rs @@ -29,6 +29,7 @@ use crate::cache::storage::{SubscriptionInfo, YangLibraryCacheError}; use netcalyx_netconf_proto::capabilities::{Capability, NetconfVersion}; use netcalyx_netconf_proto::client::{NetconfSshConnectConfig, SshAuth, SshHandler, connect}; +use netcalyx_netconf_proto::yang_module_cache::YangModuleCache; use netcalyx_netconf_proto::yang_push::filters::StreamSelectionFilterObjects; use netcalyx_netconf_proto::yang_push::subscription::{ DatastoreSelectionFilterObjects, Target, YangPushModuleVersion, @@ -86,6 +87,7 @@ struct FetchConfig { client_config: Arc, default_port: u16, timeout: std::time::Duration, + module_cache: YangModuleCache, } #[derive(Clone, Copy)] @@ -129,6 +131,7 @@ impl NetconfYangLibraryFetcher { default_port: u16, default_timeout: std::time::Duration, retry_cfg: RetryConfig, + global_module_cache: YangModuleCache, ) -> Self { Self { fetch_cfg: FetchConfig { @@ -137,6 +140,7 @@ impl NetconfYangLibraryFetcher { client_config: Arc::new(client_config), default_port, timeout: default_timeout, + module_cache: global_module_cache, }, retry_cfg, } @@ -174,7 +178,8 @@ impl NetconfYangLibraryFetcher { announce_caps, ssh_handler, Arc::clone(&cfg.client_config), - ); + ) + .with_module_cache(cfg.module_cache.clone()); let mut client = match tokio::time::timeout(cfg.timeout, connect(config)).await { Ok(Ok(c)) => c, @@ -248,7 +253,8 @@ impl NetconfYangLibraryFetcher { announce_caps, ssh_handler, Arc::clone(&cfg.client_config), - ); + ) + .with_module_cache(cfg.module_cache.clone()); // Empty subscription info returned in case of errors to keep track of peer and // subscription ID let empty = SubscriptionInfo::new_empty(peer_ip, subscription_id); From d201f712f3a46f31c49761a24376f4646053e6c2 Mon Sep 17 00:00:00 2001 From: riccardo-negri <67798955+riccardo-negri@users.noreply.github.com> Date: Mon, 10 Aug 2026 16:29:11 +0200 Subject: [PATCH 2/4] feat(netconf-proto): de-duplicate in-flight YANG module fetches Concurrent callers requesting the same (name, revision) before a fetch completes each fired their own get-schema RPC (thundering herd at startup). Add single-flight: the first caller leads and issues the RPC; others wait on its result and fall back if it fails. Keeps the RwLock (never held across await) and adds a `coalesced` counter for RPCs avoided this way. --- crates/netconf-proto/Cargo.toml | 3 +- crates/netconf-proto/src/client.rs | 74 ++++- crates/netconf-proto/src/yang_module_cache.rs | 280 ++++++++++++++++-- crates/yang-push/src/cache/actor.rs | 12 + 4 files changed, 335 insertions(+), 34 deletions(-) diff --git a/crates/netconf-proto/Cargo.toml b/crates/netconf-proto/Cargo.toml index 4590e9e7..30aff641 100644 --- a/crates/netconf-proto/Cargo.toml +++ b/crates/netconf-proto/Cargo.toml @@ -22,7 +22,7 @@ strum_macros = { workspace = true } strum = { workspace = true, features = ["derive"] } chrono = { workspace = true, default-features = false, features = ["serde"] } tokio-util = { workspace = true, default-features = false, features = ["codec"] } -tokio = { workspace = true, default-features = false } +tokio = { workspace = true, default-features = false, features = ["sync"] } tracing = { workspace = true } futures-util = { workspace = true, features = ["sink"] } russh = { workspace = true } @@ -37,3 +37,4 @@ clap = { workspace = true, features = ["derive"] } tracing-subscriber = { workspace = true, features = ["env-filter"] } anyhow = { workspace = true } serde_json = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time", "sync"] } diff --git a/crates/netconf-proto/src/client.rs b/crates/netconf-proto/src/client.rs index 9a380b24..be2a68ee 100644 --- a/crates/netconf-proto/src/client.rs +++ b/crates/netconf-proto/src/client.rs @@ -21,7 +21,7 @@ use crate::protocol::{ WellKnownOperation, WellKnownRpcResponse, YangSchemaFormat, }; use crate::xml_utils::{ParsingError, XmlDeserialize}; -use crate::yang_module_cache::YangModuleCache; +use crate::yang_module_cache::{ModuleFetch, YangModuleCache}; use crate::yang_push::SUBSCRIBED_NOTIFICATIONS_NS; use crate::yang_push::filters::Filters; use crate::yang_push::subscription::{DatastoreSelectionFilterObjects, Subscription, Target}; @@ -532,20 +532,68 @@ impl NetConfSshClient { /// populates the cache for all future calls. When `version` is `None` the /// module is fetched straight from the device on every call, bypassing the /// cache. + /// + /// Concurrent fetches of the same `(name, revision)` across sessions are + /// de-duplicated via [`YangModuleCache::begin_fetch`]: only one session + /// (the leader) issues the RPC while the others await its result. This is + /// sound because a given `(name, revision)` identifies stable content + /// regardless of which device served it ([RFC 7950] §11). + /// + /// [RFC 7950]: https://www.rfc-editor.org/rfc/rfc7950 pub async fn get_yang_module( &mut self, name: &str, version: Option<&str>, ) -> Result, NetConfSshClientError> { - if let Some(version) = version - && let Some(cached) = self.module_cache.get(name, version) - { - trace!( - "[{}] yang module cache hit for `{name}` revision {version}", - self.peer - ); - return Ok(cached); + // Unversioned modules are never cached; fetch straight from the device. + let Some(version) = version else { + return self.fetch_module_rpc(name, None).await; + }; + + // `module_cache` is a cheap-to-clone handle; clone it so the fetch below + // can borrow `&mut self` without conflicting with the cache borrow. + let cache = self.module_cache.clone(); + loop { + match cache.begin_fetch(name, version) { + ModuleFetch::Cached(cached) => { + trace!( + "[{}] yang module cache hit for `{name}` revision {version}", + self.peer + ); + return Ok(cached); + } + ModuleFetch::Wait(waiter) => { + trace!( + "[{}] awaiting in-flight fetch for `{name}` revision {version}", + self.peer + ); + if let Some(text) = waiter.wait().await { + return Ok(text); + } + // The leader failed; retry — we become a new leader or waiter. + debug!( + "[{}] leader fetch for `{name}` revision {version} failed, retrying", + self.peer + ); + } + ModuleFetch::Lead(lease) => { + // We are the leader: perform the RPC. On success publish the + // text to waiters; on error the lease drops, freeing them. + let text = self.fetch_module_rpc(name, Some(version)).await?; + lease.fulfil(Arc::clone(&text)); + return Ok(text); + } + } } + } + + /// Issue a `get-schema` RPC for `(name, version)` and return the raw module + /// text. This talks to the device directly and does not touch the cache. + async fn fetch_module_rpc( + &mut self, + name: &str, + version: Option<&str>, + ) -> Result, NetConfSshClientError> { debug!( "[{}] Getting a YANG module with name `{name}` and version {version:?}", self.peer @@ -569,13 +617,7 @@ impl NetConfSshClient { if let RpcResponse::WellKnown(WellKnownRpcResponse::YangSchema { schema }) = rpc_response { - let arc: Arc = Arc::from(schema.as_ref()); - // Only versioned modules are cached; the cache is keyed by - // `(name, revision)`. - if let Some(version) = version { - self.module_cache.insert(name, version, Arc::clone(&arc)); - } - return Ok(arc); + return Ok(Arc::from(schema.as_ref())); } else { unreachable!() } diff --git a/crates/netconf-proto/src/yang_module_cache.rs b/crates/netconf-proto/src/yang_module_cache.rs index 8796e8b7..caf5b0aa 100644 --- a/crates/netconf-proto/src/yang_module_cache.rs +++ b/crates/netconf-proto/src/yang_module_cache.rs @@ -29,18 +29,30 @@ //! //! ## Metrics //! -//! [`YangModuleCache`] exposes three plain counters via [`YangModuleCacheStats`]: +//! [`YangModuleCache`] exposes four plain counters via [`YangModuleCacheStats`]: //! //! | field | meaning | //! |-------|---------| -//! | [`YangModuleCacheStats::hits`] | `get-schema` RPC avoided (module already cached) | -//! | [`YangModuleCacheStats::misses`] | `get-schema` RPC issued (module not yet cached) | -//! | [`YangModuleCacheStats::size`] | number of distinct modules currently cached | +//! | [`YangModuleCacheStats::hits`] | `get-schema` RPC avoided (module already cached) | +//! | [`YangModuleCacheStats::misses`] | `get-schema` RPC issued (module not yet cached) | +//! | [`YangModuleCacheStats::coalesced`] | `get-schema` RPC avoided by waiting on an in-flight fetch started by another session | +//! | [`YangModuleCacheStats::size`] | number of distinct modules currently cached | //! //! These are `AtomicU64` so they can be read from any thread without holding //! the cache lock. Higher-level crates that own an OTel meter can poll them //! and record gauges / counters as needed. //! +//! ## Single-flight +//! +//! To avoid a thundering herd at collector startup — where many subscriptions +//! request the same module before any fetch has completed — the cache +//! de-duplicates **in-flight** fetches, not just completed ones. The first +//! caller for a `(name, revision)` becomes the *leader* and performs the +//! `get-schema` RPC on its own session; concurrent callers become *waiters* and +//! await the leader's result instead of issuing their own RPC. If the leader +//! fails, waiters fall back and re-race so one of them becomes a new leader. +//! This is exposed through [`YangModuleCache::begin_fetch`]. +//! //! ## References //! //! - [RFC 6022]: YANG Module for NETCONF Monitoring — defines the `get-schema` @@ -52,10 +64,24 @@ //! [RFC 7950]: https://www.rfc-editor.org/rfc/rfc7950 use std::collections::HashMap; +use std::collections::hash_map::Entry; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, RwLock}; +use tokio::sync::watch; + +/// An entry in the module cache: either a fully-fetched module text, or a +/// placeholder for a fetch that a *leader* session is currently performing. +/// Waiters clone the [`watch::Receiver`] and await the published result. +#[derive(Debug)] +enum ModuleEntry { + /// The module text is cached and ready to serve. + Ready(Arc), + /// A leader is fetching this module; the value is published here on success + /// (or the channel is closed on failure, signalling waiters to fall back). + InFlight(watch::Receiver>>), +} -type ModuleCacheInner = Arc>>>; +type ModuleCacheInner = Arc>>; /// Metrics counters exposed by [`YangModuleCache`]. #[derive(Debug, Default)] @@ -66,6 +92,9 @@ pub struct YangModuleCacheStats { /// Number of `get-schema` RPCs issued because the module was not yet /// cached. pub misses: AtomicU64, + /// Number of `get-schema` RPCs avoided by waiting on an in-flight fetch + /// started by another session (single-flight de-duplication). + pub coalesced: AtomicU64, /// Current number of distinct `(name, revision)` entries in the cache. pub size: AtomicU64, } @@ -87,6 +116,94 @@ pub struct YangModuleCache { stats: Arc, } +/// Outcome of [`YangModuleCache::begin_fetch`]: the caller's role for a given +/// `(name, revision)`. +#[derive(Debug)] +pub enum ModuleFetch { + /// The module is already cached; use this text and skip the RPC. + Cached(Arc), + /// Another session is already fetching this module; await it via + /// [`ModuleFetchWaiter::wait`] instead of issuing a duplicate RPC. + Wait(ModuleFetchWaiter), + /// No one is fetching this module yet; the caller is the leader and must + /// perform the `get-schema` RPC, then call [`ModuleFetchLease::fulfil`] + /// with the result (or drop the lease to abort, freeing waiters to retry). + Lead(ModuleFetchLease), +} + +/// Handle for a waiter to await the leader's in-flight fetch. +#[derive(Debug)] +pub struct ModuleFetchWaiter { + rx: watch::Receiver>>, + stats: Arc, +} + +impl ModuleFetchWaiter { + /// Await the leader's fetch. + /// + /// Returns `Some(text)` once the leader publishes its result (a coalesced + /// hit), or `None` if the leader failed/aborted — in which case the caller + /// should retry via [`YangModuleCache::begin_fetch`] and will become a new + /// leader or waiter. + pub async fn wait(mut self) -> Option> { + loop { + // Read the current value first so we never miss a result that was + // published before we started awaiting (watch retains the latest). + if let Some(text) = self.rx.borrow().clone() { + self.stats.coalesced.fetch_add(1, Ordering::Relaxed); + return Some(text); + } + if self.rx.changed().await.is_err() { + // Leader dropped the sender without publishing -> fall back. + return None; + } + } + } +} + +/// A lease held by the leader session while it fetches a module. +/// +/// On success the leader calls [`fulfil`](Self::fulfil) to store the text and +/// wake waiters. If the lease is dropped without fulfilment (e.g. the fetch +/// errored), the in-flight placeholder is removed so a future caller can lead, +/// and the closed watch channel signals current waiters to retry. +#[derive(Debug)] +pub struct ModuleFetchLease { + inner: ModuleCacheInner, + stats: Arc, + key: String, + tx: watch::Sender>>, + fulfilled: bool, +} + +impl ModuleFetchLease { + /// Store the fetched module text in the cache and wake all waiters. + pub fn fulfil(mut self, text: Arc) { + { + let mut map = self.inner.write().expect("yang module cache lock poisoned"); + map.insert(self.key.clone(), ModuleEntry::Ready(Arc::clone(&text))); + } + self.stats.size.fetch_add(1, Ordering::Relaxed); + // Publish to waiters; ignore send errors (no waiters is fine). + let _ = self.tx.send(Some(text)); + self.fulfilled = true; + } +} + +impl Drop for ModuleFetchLease { + fn drop(&mut self) { + if self.fulfilled { + return; + } + // Aborted fetch: remove our placeholder so the next caller re-leads. + // Dropping `tx` right after closes the watch, waking waiters to retry. + let mut map = self.inner.write().expect("yang module cache lock poisoned"); + if matches!(map.get(&self.key), Some(ModuleEntry::InFlight(_))) { + map.remove(&self.key); + } + } +} + impl YangModuleCache { pub fn new() -> Self { Self::default() @@ -96,16 +213,57 @@ impl YangModuleCache { &self.stats } + /// Begin a single-flight fetch for `(name, revision)`. + /// + /// Returns the caller's [`ModuleFetch`] role: a + /// [`Cached`](ModuleFetch::Cached) hit, a [`Wait`](ModuleFetch::Wait) + /// on another session's in-flight fetch, + /// or a [`Lead`](ModuleFetch::Lead) lease obliging the caller to fetch. + /// + /// Stats: a cache hit increments `hits`; leadership increments `misses` + /// (an RPC will be issued); waiting increments neither here — the coalesced + /// counter is bumped by [`ModuleFetchWaiter::wait`] on success. + pub fn begin_fetch(&self, name: &str, revision: &str) -> ModuleFetch { + let key = Self::make_key(name, revision); + let mut map = self.inner.write().expect("yang module cache lock poisoned"); + match map.get(&key) { + Some(ModuleEntry::Ready(text)) => { + self.stats.hits.fetch_add(1, Ordering::Relaxed); + ModuleFetch::Cached(Arc::clone(text)) + } + Some(ModuleEntry::InFlight(rx)) => ModuleFetch::Wait(ModuleFetchWaiter { + rx: rx.clone(), + stats: Arc::clone(&self.stats), + }), + None => { + self.stats.misses.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = watch::channel(None); + map.insert(key.clone(), ModuleEntry::InFlight(rx)); + ModuleFetch::Lead(ModuleFetchLease { + inner: Arc::clone(&self.inner), + stats: Arc::clone(&self.stats), + key, + tx, + fulfilled: false, + }) + } + } + } + /// Return the cached module text for `(name, revision)`, or `None` on miss. - /// Increments the appropriate stats counter. + /// Increments the appropriate stats counter. An in-flight (not yet + /// published) fetch counts as a miss. pub fn get(&self, name: &str, revision: &str) -> Option> { let key = Self::make_key(name, revision); - let result = self + let result = match self .inner .read() .expect("yang module cache lock poisoned") .get(&key) - .cloned(); + { + Some(ModuleEntry::Ready(text)) => Some(Arc::clone(text)), + Some(ModuleEntry::InFlight(_)) | None => None, + }; if result.is_some() { self.stats.hits.fetch_add(1, Ordering::Relaxed); } else { @@ -120,26 +278,25 @@ impl YangModuleCache { pub fn insert(&self, name: &str, revision: &str, text: Arc) { let key = Self::make_key(name, revision); let mut map = self.inner.write().expect("yang module cache lock poisoned"); - let prev_len = map.len(); - map.entry(key).or_insert(text); - if map.len() > prev_len { + if let Entry::Vacant(entry) = map.entry(key) { + entry.insert(ModuleEntry::Ready(text)); self.stats.size.fetch_add(1, Ordering::Relaxed); } } - /// Number of entries currently in the cache. + /// Number of ready (fully-fetched) modules currently in the cache. + /// In-flight placeholders are not counted. pub fn len(&self) -> usize { self.inner .read() .expect("yang module cache lock poisoned") - .len() + .values() + .filter(|entry| matches!(entry, ModuleEntry::Ready(_))) + .count() } pub fn is_empty(&self) -> bool { - self.inner - .read() - .expect("yang module cache lock poisoned") - .is_empty() + self.len() == 0 } fn make_key(name: &str, revision: &str) -> String { @@ -259,4 +416,93 @@ mod tests { } assert_eq!(cache.stats().size.load(Ordering::Relaxed), n_modules as u64); } + + #[tokio::test] + async fn test_single_flight_coalesces_concurrent_fetches() { + let cache = YangModuleCache::new(); + + // First caller leads. + let lease = match cache.begin_fetch("mod", "2024-01-01") { + ModuleFetch::Lead(lease) => lease, + other => panic!("first caller should lead, got {other:?}"), + }; + + // While the leader holds the lease, every concurrent caller must wait. + let n = 8; + let mut waiters = Vec::new(); + for _ in 0..n { + match cache.begin_fetch("mod", "2024-01-01") { + ModuleFetch::Wait(waiter) => { + waiters.push(tokio::spawn(async move { waiter.wait().await })); + } + other => panic!("expected wait while a fetch is in flight, got {other:?}"), + } + } + + // Publish the result; all waiters should observe it (no extra RPCs). + let text: Arc = Arc::from("module mod { }"); + lease.fulfil(Arc::clone(&text)); + + for waiter in waiters { + let got = waiter.await.expect("waiter task panicked"); + assert_eq!(got.as_deref(), Some("module mod { }")); + } + + assert_eq!(cache.stats().coalesced.load(Ordering::Relaxed), n as u64); + assert_eq!(cache.stats().misses.load(Ordering::Relaxed), 1); + assert_eq!(cache.stats().hits.load(Ordering::Relaxed), 0); + assert_eq!(cache.stats().size.load(Ordering::Relaxed), 1); + assert_eq!(cache.len(), 1); + + // A later caller now takes the fast path. + assert!(matches!( + cache.begin_fetch("mod", "2024-01-01"), + ModuleFetch::Cached(_) + )); + assert_eq!(cache.stats().hits.load(Ordering::Relaxed), 1); + } + + #[tokio::test] + async fn test_leader_failure_lets_waiters_retry() { + let cache = YangModuleCache::new(); + + let lease = match cache.begin_fetch("mod", "2024-01-01") { + ModuleFetch::Lead(lease) => lease, + other => panic!("first caller should lead, got {other:?}"), + }; + let waiter = match cache.begin_fetch("mod", "2024-01-01") { + ModuleFetch::Wait(waiter) => waiter, + other => panic!("expected wait, got {other:?}"), + }; + + // Leader aborts (e.g. its RPC failed) by dropping the lease. + drop(lease); + + // The waiter is told to fall back, and no coalesced hit is recorded. + assert_eq!(waiter.wait().await, None); + assert_eq!(cache.stats().coalesced.load(Ordering::Relaxed), 0); + assert_eq!(cache.stats().size.load(Ordering::Relaxed), 0); + + // The key is free again, so the next caller re-leads. + assert!(matches!( + cache.begin_fetch("mod", "2024-01-01"), + ModuleFetch::Lead(_) + )); + assert_eq!(cache.stats().misses.load(Ordering::Relaxed), 2); + } + + #[tokio::test] + async fn test_distinct_keys_lead_independently() { + let cache = YangModuleCache::new(); + assert!(matches!( + cache.begin_fetch("a", "2024-01-01"), + ModuleFetch::Lead(_) + )); + assert!(matches!( + cache.begin_fetch("b", "2024-01-01"), + ModuleFetch::Lead(_) + )); + // Two distinct modules -> two leaders -> two RPCs. + assert_eq!(cache.stats().misses.load(Ordering::Relaxed), 2); + } } diff --git a/crates/yang-push/src/cache/actor.rs b/crates/yang-push/src/cache/actor.rs index 45482bad..2e77beac 100644 --- a/crates/yang-push/src/cache/actor.rs +++ b/crates/yang-push/src/cache/actor.rs @@ -251,6 +251,7 @@ pub struct CachingStats { // Held here to keep the OTel callbacks registered for the lifetime of the actor. _yang_module_cache_hits: opentelemetry::metrics::ObservableCounter, _yang_module_cache_misses: opentelemetry::metrics::ObservableCounter, + _yang_module_cache_coalesced: opentelemetry::metrics::ObservableCounter, _yang_module_cache_size: opentelemetry::metrics::ObservableGauge, } @@ -312,6 +313,16 @@ impl CachingStats { }) .build() }; + let _yang_module_cache_coalesced = { + let s = Arc::clone(&stats); + meter + .u64_observable_counter("netcalyx.collector.yang_push.caching.yang_module_cache.coalesced") + .with_description("Number of get-schema RPCs avoided by waiting on an in-flight fetch started by another session") + .with_callback(move |counter| { + counter.observe(s.coalesced.load(Ordering::Relaxed), &[]); + }) + .build() + }; let _yang_module_cache_size = { let s = Arc::clone(&stats); meter @@ -336,6 +347,7 @@ impl CachingStats { device_fetch_failed, _yang_module_cache_hits, _yang_module_cache_misses, + _yang_module_cache_coalesced, _yang_module_cache_size, } } From 2334e5f6fa3a6d823d1277b1c62f80e0bec33c5c Mon Sep 17 00:00:00 2001 From: riccardo-negri <67798955+riccardo-negri@users.noreply.github.com> Date: Mon, 10 Aug 2026 16:34:27 +0200 Subject: [PATCH 3/4] refactor(netconf-proto): use a tuple key for the YANG module cache Key the cache by `(name, revision)` instead of a concatenated `name@revision` string, so the two components can't be mixed and no separator convention is baked in. --- crates/netconf-proto/src/yang_module_cache.rs | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/crates/netconf-proto/src/yang_module_cache.rs b/crates/netconf-proto/src/yang_module_cache.rs index caf5b0aa..034477a8 100644 --- a/crates/netconf-proto/src/yang_module_cache.rs +++ b/crates/netconf-proto/src/yang_module_cache.rs @@ -81,7 +81,12 @@ enum ModuleEntry { InFlight(watch::Receiver>>), } -type ModuleCacheInner = Arc>>; +/// Cache key: a `(module_name, revision)` pair. Kept as a tuple rather than a +/// concatenated string so the two components can never be mixed (e.g. a +/// name or revision containing the separator). +type ModuleCacheKey = (String, String); + +type ModuleCacheInner = Arc>>; /// Metrics counters exposed by [`YangModuleCache`]. #[derive(Debug, Default)] @@ -171,7 +176,7 @@ impl ModuleFetchWaiter { pub struct ModuleFetchLease { inner: ModuleCacheInner, stats: Arc, - key: String, + key: ModuleCacheKey, tx: watch::Sender>>, fulfilled: bool, } @@ -299,8 +304,8 @@ impl YangModuleCache { self.len() == 0 } - fn make_key(name: &str, revision: &str) -> String { - format!("{name}@{revision}") + fn make_key(name: &str, revision: &str) -> ModuleCacheKey { + (name.to_owned(), revision.to_owned()) } } From dca72bad6f44a46bea44212dacb14eaf2bceae06 Mon Sep 17 00:00:00 2001 From: riccardo-negri <67798955+riccardo-negri@users.noreply.github.com> Date: Mon, 10 Aug 2026 16:53:43 +0200 Subject: [PATCH 4/4] refactor(netconf-proto): tidy up YANG module cache internals --- crates/netconf-proto/src/client.rs | 4 +- crates/netconf-proto/src/yang_module_cache.rs | 46 +++++++++++++------ 2 files changed, 33 insertions(+), 17 deletions(-) diff --git a/crates/netconf-proto/src/client.rs b/crates/netconf-proto/src/client.rs index be2a68ee..c607f74e 100644 --- a/crates/netconf-proto/src/client.rs +++ b/crates/netconf-proto/src/client.rs @@ -17,8 +17,8 @@ use crate::capabilities::{Capability, NetconfVersion}; use crate::codec::{SshCodec, SshCodecError}; use crate::protocol::{ - Filter, Hello, NetConfMessage, Rpc, RpcOperation, RpcReply, RpcReplyContent, RpcResponse, - WellKnownOperation, WellKnownRpcResponse, YangSchemaFormat, + Filter, GetSchemaFormat, Hello, NetConfMessage, Rpc, RpcOperation, RpcReply, RpcReplyContent, + RpcResponse, WellKnownOperation, WellKnownRpcResponse, }; use crate::xml_utils::{ParsingError, XmlDeserialize}; use crate::yang_module_cache::{ModuleFetch, YangModuleCache}; diff --git a/crates/netconf-proto/src/yang_module_cache.rs b/crates/netconf-proto/src/yang_module_cache.rs index 034477a8..7b2fcaec 100644 --- a/crates/netconf-proto/src/yang_module_cache.rs +++ b/crates/netconf-proto/src/yang_module_cache.rs @@ -231,19 +231,22 @@ impl YangModuleCache { pub fn begin_fetch(&self, name: &str, revision: &str) -> ModuleFetch { let key = Self::make_key(name, revision); let mut map = self.inner.write().expect("yang module cache lock poisoned"); - match map.get(&key) { - Some(ModuleEntry::Ready(text)) => { - self.stats.hits.fetch_add(1, Ordering::Relaxed); - ModuleFetch::Cached(Arc::clone(text)) - } - Some(ModuleEntry::InFlight(rx)) => ModuleFetch::Wait(ModuleFetchWaiter { - rx: rx.clone(), - stats: Arc::clone(&self.stats), - }), - None => { + match map.entry(key) { + Entry::Occupied(entry) => match entry.get() { + ModuleEntry::Ready(text) => { + self.stats.hits.fetch_add(1, Ordering::Relaxed); + ModuleFetch::Cached(Arc::clone(text)) + } + ModuleEntry::InFlight(rx) => ModuleFetch::Wait(ModuleFetchWaiter { + rx: rx.clone(), + stats: Arc::clone(&self.stats), + }), + }, + Entry::Vacant(entry) => { self.stats.misses.fetch_add(1, Ordering::Relaxed); let (tx, rx) = watch::channel(None); - map.insert(key.clone(), ModuleEntry::InFlight(rx)); + let key = entry.key().clone(); + entry.insert(ModuleEntry::InFlight(rx)); ModuleFetch::Lead(ModuleFetchLease { inner: Arc::clone(&self.inner), stats: Arc::clone(&self.stats), @@ -258,7 +261,13 @@ impl YangModuleCache { /// Return the cached module text for `(name, revision)`, or `None` on miss. /// Increments the appropriate stats counter. An in-flight (not yet /// published) fetch counts as a miss. - pub fn get(&self, name: &str, revision: &str) -> Option> { + /// + /// Test-only: production code goes through + /// [`begin_fetch`](Self::begin_fetch) so that concurrent fetches are + /// de-duplicated. A plain `get` would bypass single-flight and re-issue + /// redundant `get-schema` RPCs. + #[cfg(test)] + fn get(&self, name: &str, revision: &str) -> Option> { let key = Self::make_key(name, revision); let result = match self .inner @@ -280,7 +289,12 @@ impl YangModuleCache { /// Insert a module text. First writer wins: if `(name, revision)` is /// already present the call is a no-op. This is safe because identical /// `(name, revision)` always has identical content per the YANG spec. - pub fn insert(&self, name: &str, revision: &str, text: Arc) { + /// + /// Test-only: production code publishes through + /// [`ModuleFetchLease::fulfil`], which is driven by + /// [`begin_fetch`](Self::begin_fetch)'s single-flight protocol. + #[cfg(test)] + fn insert(&self, name: &str, revision: &str, text: Arc) { let key = Self::make_key(name, revision); let mut map = self.inner.write().expect("yang module cache lock poisoned"); if let Entry::Vacant(entry) = map.entry(key) { @@ -291,7 +305,8 @@ impl YangModuleCache { /// Number of ready (fully-fetched) modules currently in the cache. /// In-flight placeholders are not counted. - pub fn len(&self) -> usize { + #[cfg(test)] + fn len(&self) -> usize { self.inner .read() .expect("yang module cache lock poisoned") @@ -300,7 +315,8 @@ impl YangModuleCache { .count() } - pub fn is_empty(&self) -> bool { + #[cfg(test)] + fn is_empty(&self) -> bool { self.len() == 0 }