Fix deserialization when sample_freq values contain a % symbol

Co-authored-by: Benjamin Sparks <b.sparks@alugha.com>

Author: bengsparks

async-nats/src/jetstream/consumer/mod.rs

@@ -273,7 +273,7 @@ pub struct Config {
     /// What percentage of acknowledgments should be samples for observability, 0-100
     #[serde(
         rename = "sample_freq",
-        with = "from_str",
+        with = "sample_freq_deser",
         default,
         skip_serializing_if = "is_default"
     )]
@@ -433,15 +433,25 @@ fn is_default<T: Default + Eq>(t: &T) -> bool {
     t == &T::default()
 }
 
-pub(crate) mod from_str {
+pub(crate) mod sample_freq_deser {
     pub(crate) fn deserialize<'de, T, D>(deserializer: D) -> Result<T, D::Error>
     where
         T: std::str::FromStr,
         T::Err: std::fmt::Display,
         D: serde::Deserializer<'de>,
     {
         let s = <String as serde::Deserialize>::deserialize(deserializer)?;
-        T::from_str(&s).map_err(serde::de::Error::custom)
+
+        let mut spliterator = s.split('%');
+        match (spliterator.next(), spliterator.next()) {
+            // No percentage occurred, parse as number
+            (Some(number), None) => T::from_str(number).map_err(serde::de::Error::custom),
+            // A percentage sign occurred right at the end
+            (Some(number), Some("")) => T::from_str(number).map_err(serde::de::Error::custom),
+            _ => Err(serde::de::Error::custom(format!(
+                "Malformed sample frequency: {s}"
+            ))),
+        }
     }
 
     pub(crate) fn serialize<T, S>(value: &T, serializer: S) -> Result<S::Ok, S::Error>

async-nats/src/jetstream/consumer/pull.rs

@@ -605,7 +605,7 @@ pub struct OrderedConfig {
     /// What percentage of acknowledgments should be samples for observability, 0-100
     #[serde(
         rename = "sample_freq",
-        with = "super::from_str",
+        with = "super::sample_freq_deser",
         default,
         skip_serializing_if = "is_default"
     )]
@@ -2051,7 +2051,7 @@ pub struct Config {
     /// What percentage of acknowledgments should be samples for observability, 0-100
     #[serde(
         rename = "sample_freq",
-        with = "super::from_str",
+        with = "super::sample_freq_deser",
         default,
         skip_serializing_if = "is_default"
     )]

async-nats/src/jetstream/consumer/push.rs

@@ -236,7 +236,7 @@ pub struct Config {
     /// What percentage of acknowledgments should be samples for observability, 0-100
     #[serde(
         rename = "sample_freq",
-        with = "super::from_str",
+        with = "super::sample_freq_deser",
         default,
         skip_serializing_if = "is_default"
     )]
@@ -389,7 +389,7 @@ pub struct OrderedConfig {
     /// What percentage of acknowledgments should be samples for observability, 0-100
     #[serde(
         rename = "sample_freq",
-        with = "super::from_str",
+        with = "super::sample_freq_deser",
         default,
         skip_serializing_if = "is_default"
     )]

async-nats/tests/jetstream_tests.rs

@@ -2617,6 +2617,31 @@ mod jetstream {
 
             assert_eq!(100, consumer.cached_info().config.sample_frequency);
         }
+
+        // `sample_frequency` can contain a %, which occurs in the wild
+        // when managing NATS resources using Terraform / OpenTofu's NATS JetStream provider.
+        {
+            let stream = &stream.cached_info().config.name;
+            let consumer = serde_json::json!({
+                "stream_name": stream,
+                "config": {
+                    "name": "consumer",
+                    "sample_freq": "10%",
+                },
+            });
+
+            let response: Response<Info> = js
+                .request(format!("CONSUMER.CREATE.{}", stream), &consumer)
+                .await
+                .unwrap();
+
+            match response {
+                Response::Ok(info) => {
+                    assert_eq!(info.config.sample_frequency, 10);
+                }
+                Response::Err { error } => panic!("expected ok response, got: {:?}", error),
+            }
+        }
     }
 
     #[tokio::test]