Add get_with_version for atomic content+version reads

Summary:
Adds `ConfigHandle::get_with_version(&self) -> (Arc<T>, Option<ConfigVersionInfo>)`, an accessor that returns the config contents together with the version metadata (`version` string + `mod_time`) they were parsed from, read atomically from one snapshot.

**The skew problem.** Today a `RegisteredConfigEntity`'s deserialized contents live in a tokio watch channel while `version`/`mod_time` live in a separate `RwLock<CachedConfigEntity>`. `refresh()` sends the new contents to the channel *before* committing the new version to the lock, so calling `get()` and then reading the version separately can pair new contents with the old version (or vice versa) for one refresh cycle.

**The fix.** `CachedConfigEntity` now also carries the deserialized `Arc<T>`, committed in the same write-lock critical section as `version`/`mod_time`; `get_with_version` reads all three under a single read-lock acquisition, so the returned pair always corresponds. `get()` semantics and the watch-channel behavior are unchanged (the extra `Arc` in the lock is just another reference to the same allocation).

**The Fixed contract.** Handles backed by static configs (`from_json`, `default`, `From<T>`) have no version, so the version half is `Option`-typed and they return `(contents, None)` — never a sentinel string. `ConfigVersionInfo` is `#[non_exhaustive]` so future fields (e.g. mutation id) can be added without breaking callers.

Also adds `TestSource::insert_config_with_version` so tests can exercise non-empty versions (`insert_config` previously hardcoded `version: String::new()` and now delegates).

**Consumer.** This is Diff A1 of the Task 32 stack (Mononoke: decouple startup from legacy tier blobs; design doc: configerator `source/scm/mononoke/docs/plans/2026-08-17-task32-blob-decouple-design-final.md`). Mononoke's per-repo config provenance (Diff B) will call this accessor at parse time and log the config version to scuba.

Reviewed By: YousefSalama

Differential Revision: D116321758

fbshipit-source-id: fab12e34e2f2da92da257d62b1a5fbf749685e9d

Author: RajivTS

shed/cached_config/src/handle.rs

@@ -17,8 +17,24 @@ use serde::de::DeserializeOwned;
 use serde_json::from_str;
 use tokio::sync::watch::Receiver;
 
+use crate::ModificationTime;
 use crate::refreshable_entities::RegisteredConfigEntity;
 
+/// Versioning metadata for a config snapshot, as reported by the config
+/// source (e.g. configerator's Entity version). For configerator-backed
+/// sources the version uniquely identifies the config snapshot; other
+/// `Source` impls (e.g. `TestSource` without explicit versions) may reuse
+/// versions across changing contents — pair with `mod_time` when identity
+/// matters.
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct ConfigVersionInfo {
+    /// Version of the config as reported by the config source
+    pub version: String,
+    /// Modification time of the config, e.g. file modification time
+    pub mod_time: ModificationTime,
+}
+
 /// A configuration handle, with self-refresh and wait-on-update if obtained
 /// from a `ConfigStore`. If your type `T` implements `Default`, then this
 /// will implement `Default` using a fixed config matching `T`'s default
@@ -80,6 +96,22 @@ where
         }
     }
 
+    /// Fetch the current version of the config referred to by this handle
+    /// together with its versioning metadata. Both halves of the returned
+    /// tuple are read atomically from the same snapshot, so the version info
+    /// always corresponds to the returned contents, even if the config is
+    /// concurrently refreshed. Fixed (static) configs, e.g. obtained via
+    /// `from_json` or `default`, have no version and return `None`.
+    pub fn get_with_version(&self) -> (Arc<T>, Option<ConfigVersionInfo>) {
+        match &self.inner {
+            ConfigHandleImpl::Registered(handle) => {
+                let (contents, version_info) = handle.get_with_version();
+                (contents, Some(version_info))
+            }
+            ConfigHandleImpl::Fixed(contents) => (contents.clone(), None),
+        }
+    }
+
     /// Method that returns a config update watcher that observes changes
     /// that are applied to the underlying config. Requesting a config watcher
     /// for static config (e.g. sourced via static JSON file) results in an error.

shed/cached_config/src/lib.rs

@@ -34,6 +34,7 @@ use bytes::Bytes;
 use chrono::NaiveDateTime;
 pub use handle::ConfigHandle;
 pub use handle::ConfigUpdateWatcher;
+pub use handle::ConfigVersionInfo;
 pub use store::ConfigStore;
 pub use test_source::TestSource;
 

shed/cached_config/src/refreshable_entities.rs

@@ -20,6 +20,7 @@ use tokio::sync::watch::channel;
 
 use crate::Entity;
 use crate::ModificationTime;
+use crate::handle::ConfigVersionInfo;
 
 // Type-erasure trick. I don't actually care about T for RegisteredConfigEntity,
 /// so hide it via a trait object
@@ -30,16 +31,21 @@ pub(crate) trait Refreshable {
 
 /// The type contained in a `ConfigHandle` when it's obtained from a `ConfigStore`
 pub(crate) struct RegisteredConfigEntity<T> {
-    contents: RwLock<CachedConfigEntity>,
+    contents: RwLock<CachedConfigEntity<T>>,
     path: String,
     deserializer: fn(Bytes) -> Result<T>,
     update_sender: RwLock<Sender<Arc<T>>>,
     update_receiver: RwLock<Receiver<Arc<T>>>,
 }
 
-struct CachedConfigEntity {
+/// A single config snapshot: the deserialized contents together with the
+/// version metadata they were parsed from. All fields are committed in the
+/// same write-lock critical section during `refresh`, so readers holding the
+/// read lock always observe contents and version that correspond.
+struct CachedConfigEntity<T> {
     mod_time: ModificationTime,
     version: String,
+    contents: Arc<T>,
 }
 
 impl<T> RegisteredConfigEntity<T>
@@ -57,10 +63,14 @@ where
             contents,
         } = entity;
         let contents = Arc::new(deserializer(contents.unwrap_or_else(Bytes::new))?);
-        let (update_sender, update_receiver) = channel(contents);
+        let (update_sender, update_receiver) = channel(contents.clone());
 
         Ok(Self {
-            contents: RwLock::new(CachedConfigEntity { mod_time, version }),
+            contents: RwLock::new(CachedConfigEntity {
+                mod_time,
+                version,
+                contents,
+            }),
             path,
             deserializer,
             update_sender: RwLock::new(update_sender),
@@ -76,6 +86,21 @@ where
             .clone()
     }
 
+    /// Get the current contents together with the version metadata they were
+    /// parsed from. Both are read under a single read-lock acquisition, so
+    /// they are guaranteed to correspond to the same config snapshot even if
+    /// a refresh is in flight.
+    pub(crate) fn get_with_version(&self) -> (Arc<T>, ConfigVersionInfo) {
+        let locked = self.contents.read().expect("lock poisoned");
+        (
+            locked.contents.clone(),
+            ConfigVersionInfo {
+                version: locked.version.clone(),
+                mod_time: locked.mod_time.clone(),
+            },
+        )
+    }
+
     pub(crate) fn update_receiver(&self) -> Receiver<Arc<T>> {
         self.update_receiver.read().expect("lock poisoned").clone()
     }
@@ -98,7 +123,11 @@ where
         if has_changed {
             let contents = Arc::new((self.deserializer)(entity.contents.unwrap_or_default())?);
             let update_sender = self.update_sender.write().expect("lock poisoned");
-            if update_sender.send(contents).is_err() {
+            // Deliberate ordering: the watch channel is updated before the
+            // snapshot lock below is committed, so watchers/get() can briefly
+            // see newer contents than get_with_version(). Each accessor family
+            // is self-consistent; do not "fix" the order.
+            if update_sender.send(contents.clone()).is_err() {
                 bail!(
                     "No subscriber for config updates at path {}",
                     self.get_path()
@@ -109,6 +138,7 @@ where
                 *locked = CachedConfigEntity {
                     mod_time: entity.mod_time,
                     version: entity.version,
+                    contents,
                 };
                 Ok(true)
             }

shed/cached_config/src/test_source.rs

@@ -55,13 +55,25 @@ impl TestSource {
 
     /// Insert config value into the `TestSource`, overwriting existing one
     pub fn insert_config(&self, key: &str, contents: &str, mod_time: ModificationTime) {
+        self.insert_config_with_version(key, contents, mod_time, "");
+    }
+
+    /// Insert config value with an explicit version into the `TestSource`,
+    /// overwriting existing one
+    pub fn insert_config_with_version(
+        &self,
+        key: &str,
+        contents: &str,
+        mod_time: ModificationTime,
+        version: &str,
+    ) {
         let mut map = self.path_to_config.lock().expect("poisoned lock");
         map.insert(
             key.to_owned(),
             Entity {
                 contents: Some(Bytes::copy_from_slice(contents.as_bytes())),
                 mod_time,
-                version: String::new(),
+                version: version.to_owned(),
             },
         );
     }

shed/cached_config/src/tests.rs

@@ -109,6 +109,121 @@ fn test_config_store() {
     assert_eq!(*raw_handle.get(), r#"{ "value": 11 }"#);
 }
 
+#[test]
+fn test_get_with_version_registered() {
+    let test_source = {
+        let test_source = TestSource::new();
+        test_source.insert_config_with_version(
+            "some1",
+            r#"{ "value": 1 }"#,
+            ModificationTime::UnixTimestamp(1),
+            "v1",
+        );
+        Arc::new(test_source)
+    };
+
+    // No poll interval: refreshes happen only via force_update_configs, so the
+    // test is deterministic.
+    let store = ConfigStore::new(test_source.clone(), None::<Duration>, None);
+
+    let handle = get_test_handle(&store, "some1").expect("Failed to get handle");
+
+    let (contents, info) = handle.get_with_version();
+    assert_eq!(
+        *contents,
+        TestConfig { value: 1 },
+        "contents should match the initially inserted config"
+    );
+    let info = info.expect("registered handles should always report version info");
+    assert_eq!(
+        info.version, "v1",
+        "version should match the inserted config's version"
+    );
+    assert_eq!(
+        info.mod_time,
+        ModificationTime::UnixTimestamp(1),
+        "mod_time should match the inserted config's mod_time"
+    );
+
+    // Update the config and refresh; contents and version from ONE call must
+    // correspond to the same snapshot.
+    test_source.insert_config_with_version(
+        "some1",
+        r#"{ "value": 2 }"#,
+        ModificationTime::UnixTimestamp(2),
+        "v2",
+    );
+    test_source.insert_to_refresh("some1".to_owned());
+    store.force_update_configs();
+
+    let (contents, info) = handle.get_with_version();
+    let info = info.expect("registered handles should always report version info");
+    assert_eq!(
+        *contents,
+        TestConfig { value: 2 },
+        "contents should reflect the refreshed config"
+    );
+    assert_eq!(
+        info.version, "v2",
+        "version must correspond to the contents returned by the same call"
+    );
+    assert_eq!(
+        info.mod_time,
+        ModificationTime::UnixTimestamp(2),
+        "mod_time must correspond to the contents returned by the same call"
+    );
+
+    // get() must be unaffected by the new accessor
+    assert_eq!(
+        *handle.get(),
+        TestConfig { value: 2 },
+        "get() should still return the refreshed config"
+    );
+}
+
+#[test]
+fn test_get_with_version_legacy_insert_config() {
+    let test_source = {
+        let test_source = TestSource::new();
+        test_source.insert_config(
+            "legacy",
+            r#"{ "value": 3 }"#,
+            ModificationTime::UnixTimestamp(1),
+        );
+        Arc::new(test_source)
+    };
+    let store = ConfigStore::new(test_source, None::<Duration>, None);
+
+    let handle = get_test_handle(&store, "legacy").expect("Failed to get handle");
+    let (contents, info) = handle.get_with_version();
+    assert_eq!(
+        *contents,
+        TestConfig { value: 3 },
+        "contents should match the inserted config"
+    );
+    let info = info.expect("registered handles should always report version info");
+    assert_eq!(
+        info.version, "",
+        "legacy insert_config must keep reporting an empty version for backward compatibility"
+    );
+}
+
+#[test]
+fn test_get_with_version_fixed() {
+    let handle = ConfigHandle::<TestConfig>::from_json(r#"{ "value": 44 }"#)
+        .expect("failed to deserialize json");
+    let (contents, info) = handle.get_with_version();
+    assert_eq!(
+        *contents,
+        TestConfig { value: 44 },
+        "contents should match the fixed JSON config"
+    );
+    assert!(
+        info.is_none(),
+        "fixed (from_json/default) handles must return None version info, not a sentinel"
+    );
+}
+
 #[test]
 fn test_config_handle_from_json() {
     let result = ConfigHandle::<TestConfig>::from_json(r#"{ "value": 44 }"#)