From 97161dbd0b407eacfc47cc5d4fffe00154b74e8c Mon Sep 17 00:00:00 2001 From: Nathaniel Ford Date: Wed, 17 Jun 2026 20:09:48 +0000 Subject: [PATCH 1/6] Implement Service configuration deserialization. --- grpc/src/client/channel.rs | 5 +- grpc/src/client/service_config.rs | 38 --- grpc/src/client/service_config/json.rs | 266 +++++++++++++++++++++ grpc/src/client/service_config/mod.rs | 310 +++++++++++++++++++++++++ grpc/src/inmemory/mod.rs | 11 +- 5 files changed, 586 insertions(+), 44 deletions(-) delete mode 100644 grpc/src/client/service_config.rs create mode 100644 grpc/src/client/service_config/json.rs create mode 100644 grpc/src/client/service_config/mod.rs diff --git a/grpc/src/client/channel.rs b/grpc/src/client/channel.rs index 45dccf03a..4caa1db75 100644 --- a/grpc/src/client/channel.rs +++ b/grpc/src/client/channel.rs @@ -68,7 +68,6 @@ use crate::client::name_resolution::Target; use crate::client::name_resolution::dns; use crate::client::name_resolution::global_registry; use crate::client::name_resolution::{self}; -use crate::client::service_config::LbPolicyType; use crate::client::service_config::ServiceConfig; use crate::client::stream_util::FailingRecvStream; use crate::client::subchannel::InternalSubchannel; @@ -464,9 +463,9 @@ impl name_resolution::ChannelController for ResolverChannelController { fn update(&mut self, update: ResolverUpdate) -> Result<(), String> { let json_config = if let Ok(Some(service_config)) = update.service_config.as_ref() && service_config - .load_balancing_policy + .load_balancing_config .as_ref() - .is_some_and(|p| *p == LbPolicyType::RoundRobin) + .is_some_and(|lbc| lbc.iter().any(|c| c.name == "round_robin")) { json!([{round_robin::POLICY_NAME: {}}]) } else { diff --git a/grpc/src/client/service_config.rs b/grpc/src/client/service_config.rs deleted file mode 100644 index c9c000899..000000000 --- a/grpc/src/client/service_config.rs +++ /dev/null @@ -1,38 +0,0 @@ -/* - * - * Copyright 2025 gRPC authors. - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to - * deal in the Software without restriction, including without limitation the - * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - * sell copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - * IN THE SOFTWARE. - * - */ - -/// An in-memory representation of a service config, usually provided to gRPC as -/// a JSON object. -// TODO: complete this and ensure it meets all our requirements. -#[derive(Debug, Default, Clone)] -pub struct ServiceConfig { - pub(crate) load_balancing_policy: Option, -} - -#[derive(Debug, Default, Clone, PartialEq, Eq)] -pub enum LbPolicyType { - #[default] - PickFirst, - RoundRobin, -} diff --git a/grpc/src/client/service_config/json.rs b/grpc/src/client/service_config/json.rs new file mode 100644 index 000000000..9771fb494 --- /dev/null +++ b/grpc/src/client/service_config/json.rs @@ -0,0 +1,266 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use serde::Deserialize; +use std::time::Duration; + +pub(crate) fn parse_duration(s: &str) -> Result { + if !s.ends_with('s') { + return Err("duration string must end with 's'".to_string()); + } + let s = &s[..s.len() - 1]; // strip 's'. + let mut parts = s.splitn(2, '.'); + let secs_str = parts + .next() + .ok_or_else(|| "empty duration string".to_string())?; + let secs: u64 = secs_str + .parse() + .map_err(|e| format!("failed to parse seconds: {}", e))?; + + let nanos = if let Some(fraction_str) = parts.next() { + if fraction_str.is_empty() { + return Err("empty fraction part".to_string()); + } + if fraction_str.len() > 9 { + return Err("fraction part has more than 9 digits".to_string()); + } + let fraction_val: u32 = fraction_str + .parse() + .map_err(|e| format!("failed to parse fraction: {}", e))?; + let pad = 9 - fraction_str.len(); + fraction_val * 10u32.pow(pad as u32) + } else { + 0 + }; + + Ok(Duration::new(secs, nanos)) +} + +pub(crate) fn deserialize_duration<'de, D>(deserializer: D) -> Result +where + D: serde::Deserializer<'de>, +{ + let s = String::deserialize(deserializer)?; + parse_duration(&s).map_err(serde::de::Error::custom) +} + +pub(crate) fn deserialize_duration_opt<'de, D>( + deserializer: D, +) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + struct OptVisitor; + impl<'de> serde::de::Visitor<'de> for OptVisitor { + type Value = Option; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("a string representing a duration or null") + } + + fn visit_none(self) -> Result + where + E: serde::de::Error, + { + Ok(None) + } + + fn visit_some(self, deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + parse_duration(&s) + .map(Some) + .map_err(serde::de::Error::custom) + } + } + deserializer.deserialize_option(OptVisitor) +} + +pub(crate) fn deserialize_int64<'de, D>(deserializer: D) -> Result +where + D: serde::Deserializer<'de>, +{ + struct Visitor; + impl<'de> serde::de::Visitor<'de> for Visitor { + type Value = i64; + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("an i64 or a string representing an i64") + } + fn visit_i64(self, v: i64) -> Result + where + E: serde::de::Error, + { + Ok(v) + } + fn visit_u64(self, v: u64) -> Result + where + E: serde::de::Error, + { + v.try_into().map_err(serde::de::Error::custom) + } + fn visit_str(self, v: &str) -> Result + where + E: serde::de::Error, + { + v.parse().map_err(serde::de::Error::custom) + } + } + deserializer.deserialize_any(Visitor) +} + +pub(crate) fn deserialize_uint32_opt<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + struct OptVisitor; + impl<'de> serde::de::Visitor<'de> for OptVisitor { + type Value = Option; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("a u32 or a string representing a u32 or null") + } + + fn visit_none(self) -> Result + where + E: serde::de::Error, + { + Ok(None) + } + + fn visit_u64(self, v: u64) -> Result + where + E: serde::de::Error, + { + v.try_into().map(Some).map_err(serde::de::Error::custom) + } + + fn visit_str(self, v: &str) -> Result + where + E: serde::de::Error, + { + v.parse().map(Some).map_err(serde::de::Error::custom) + } + + fn visit_some(self, deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_any(self) + } + } + deserializer.deserialize_option(OptVisitor) +} + +#[cfg(test)] +mod test { + use super::*; + use serde_json::json; + + #[test] + fn test_parse_duration() { + assert_eq!(parse_duration("1s").unwrap(), Duration::from_secs(1)); + assert_eq!(parse_duration("0s").unwrap(), Duration::from_secs(0)); + assert_eq!( + parse_duration("1.5s").unwrap(), + Duration::new(1, 500_000_000) + ); + assert_eq!(parse_duration("0.000000001s").unwrap(), Duration::new(0, 1)); + assert_eq!(parse_duration("1.0000001s").unwrap(), Duration::new(1, 100)); + assert_eq!( + parse_duration("1.0001s").unwrap(), + Duration::new(1, 100_000) + ); + + assert!(parse_duration("").is_err()); + assert!(parse_duration("1").is_err()); + assert!(parse_duration("1.s").is_err()); + assert!(parse_duration("1.0000000001s").is_err()); + assert!(parse_duration("as").is_err()); + } + + #[test] + fn test_deserialize_duration() { + #[derive(Deserialize, Debug, PartialEq)] + struct TestStruct { + #[serde(deserialize_with = "deserialize_duration")] + d: Duration, + } + + let val: TestStruct = serde_json::from_value(json!({ "d": "1.5s" })).unwrap(); + assert_eq!(val.d, Duration::new(1, 500_000_000)); + + let res: Result = serde_json::from_value(json!({ "d": "1" })); + assert!(res.is_err()); + } + + #[test] + fn test_deserialize_duration_opt() { + #[derive(Deserialize, Debug, PartialEq)] + struct TestStruct { + #[serde(default, deserialize_with = "deserialize_duration_opt")] + d: Option, + } + + let val: TestStruct = serde_json::from_value(json!({ "d": "1.5s" })).unwrap(); + assert_eq!(val.d, Some(Duration::new(1, 500_000_000))); + + let val: TestStruct = serde_json::from_value(json!({ "d": null })).unwrap(); + assert_eq!(val.d, None); + + let val: TestStruct = serde_json::from_value(json!({})).unwrap(); + assert_eq!(val.d, None); + + let res: Result = serde_json::from_value(json!({ "d": "invalid" })); + assert!(res.is_err()); + } + + #[test] + fn test_deserialize_uint32_opt() { + #[derive(Deserialize, Debug, PartialEq)] + struct TestStruct { + #[serde(default, deserialize_with = "deserialize_uint32_opt")] + val: Option, + } + + let val: TestStruct = serde_json::from_value(json!({ "val": 123 })).unwrap(); + assert_eq!(val.val, Some(123)); + + let val: TestStruct = serde_json::from_value(json!({ "val": "456" })).unwrap(); + assert_eq!(val.val, Some(456)); + + let val: TestStruct = serde_json::from_value(json!({ "val": null })).unwrap(); + assert_eq!(val.val, None); + + let val: TestStruct = serde_json::from_value(json!({})).unwrap(); + assert_eq!(val.val, None); + + let res: Result = serde_json::from_value(json!({ "val": "invalid" })); + assert!(res.is_err()); + + let res: Result = serde_json::from_value(json!({ "val": -1 })); + assert!(res.is_err()); + } +} diff --git a/grpc/src/client/service_config/mod.rs b/grpc/src/client/service_config/mod.rs new file mode 100644 index 000000000..a7c52fc7f --- /dev/null +++ b/grpc/src/client/service_config/mod.rs @@ -0,0 +1,310 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use serde::{Deserialize, Serialize}; +use std::time::Duration; + +pub(crate) mod json; + +/// An in-memory representation of a service config, provided to gRPC as a JSON object. +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct ServiceConfig { + /// Ordered list of LB policies. The first supported one is used. + pub load_balancing_config: Option>, + /// Per-method configuration overrides. + pub method_config: Option>, + /// Global retry throttling parameters. + pub retry_throttling: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct MethodConfig { + /// List of methods this config applies to. + pub name: Vec, + #[serde(default, deserialize_with = "json::deserialize_duration_opt")] + pub timeout: Option, + pub retry_policy: Option, + #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] + pub max_request_message_bytes: Option, + #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] + pub max_response_message_bytes: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +pub struct MethodName { + /// e.g., "grpc.examples.echo.Echo" or "grpc.examples.echo.Echo/Echo". + /// The service name (fully qualified). + pub service: String, + /// If None, applies to all methods in the service. + pub method: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct RetryThrottlingPolicy { + pub max_tokens: u32, + pub token_ratio: f32, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct RetryPolicy { + pub max_attempts: u32, + #[serde(deserialize_with = "json::deserialize_duration")] + pub initial_backoff: Duration, + #[serde(deserialize_with = "json::deserialize_duration")] + pub max_backoff: Duration, + pub backoff_multiplier: f32, + pub retryable_status_codes: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LoadBalancingConfig { + pub name: String, + pub config: serde_json::Value, +} + +impl<'de> Deserialize<'de> for LoadBalancingConfig { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + struct Visitor; + impl<'de> serde::de::Visitor<'de> for Visitor { + type Value = LoadBalancingConfig; + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str( + "a map with a single key-value pair representing a load balancing policy", + ) + } + fn visit_map(self, mut map: A) -> Result + where + A: serde::de::MapAccess<'de>, + { + if let Some(name) = map.next_key::()? { + let config = map.next_value::()?; + if map.next_key::()?.is_some() { + return Err(serde::de::Error::custom("map has more than one key")); + } + Ok(LoadBalancingConfig { name, config }) + } else { + Err(serde::de::Error::custom("map is empty")) + } + } + } + deserializer.deserialize_map(Visitor) + } +} + +impl Serialize for LoadBalancingConfig { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + use serde::ser::SerializeMap; + let mut map = serializer.serialize_map(Some(1))?; + map.serialize_entry(&self.name, &self.config)?; + map.end() + } +} + +impl ServiceConfig { + /// Parses a service configuration from a JSON string. + pub fn parse(config_json: &str) -> Result { + let config: Self = serde_json::from_str(config_json) + .map_err(|e| format!("failed to deserialize service config JSON: {}", e))?; + config.validate()?; + Ok(config) + } + + fn validate(&self) -> Result<(), String> { + if let Some(ref method_configs) = self.method_config { + for (i, mc) in method_configs.iter().enumerate() { + if mc.name.is_empty() { + return Err(format!("method_config[{}] has no names defined", i)); + } + for (j, name) in mc.name.iter().enumerate() { + if name.service.is_empty() { + return Err(format!( + "method_config[{}].name[{}] has empty service name", + i, j + )); + } + } + if let Some(ref rp) = mc.retry_policy { + if rp.max_attempts <= 1 { + return Err(format!( + "method_config[{}].retry_policy.max_attempts must be > 1", + i + )); + } + if rp.initial_backoff.as_nanos() == 0 { + return Err(format!( + "method_config[{}].retry_policy.initial_backoff must be > 0", + i + )); + } + if rp.max_backoff.as_nanos() == 0 { + return Err(format!( + "method_config[{}].retry_policy.max_backoff must be > 0", + i + )); + } + if rp.backoff_multiplier <= 0.0 { + return Err(format!( + "method_config[{}].retry_policy.backoff_multiplier must be > 0", + i + )); + } + } + } + } + + if let Some(ref rt) = self.retry_throttling { + if rt.max_tokens == 0 || rt.max_tokens > 1000 { + return Err("retry_throttling.max_tokens must be between 1 and 1000".to_string()); + } + if rt.token_ratio <= 0.0 { + return Err("retry_throttling.token_ratio must be > 0".to_string()); + } + } + + Ok(()) + } +} + +#[cfg(test)] +mod test { + use super::*; + use serde_json::json; + + #[test] + fn test_valid_service_config_parsing() { + let json_data = json!({ + "loadBalancingConfig": [ + { "pick_first": { "shuffleAddressList": true } }, + { "round_robin": {} } + ], + "methodConfig": [ + { + "name": [ + { "service": "grpc.examples.echo.Echo", "method": "Echo" }, + { "service": "grpc.examples.echo.Echo2" } + ], + "timeout": "1.5s", + "retryPolicy": { + "maxAttempts": 3, + "initialBackoff": "0.1s", + "maxBackoff": "1s", + "backoffMultiplier": 2.0, + "retryableStatusCodes": ["UNAVAILABLE", "INTERNAL"] + }, + "maxRequestMessageBytes": 1024, + "maxResponseMessageBytes": "2048" + } + ], + "retryThrottling": { + "maxTokens": 100, + "tokenRatio": 0.1 + } + }); + + let sc = ServiceConfig::parse(&json_data.to_string()).unwrap(); + + // Verify Load Balancing Config. + let lb_configs = sc.load_balancing_config.unwrap(); + assert_eq!(lb_configs.len(), 2); + assert_eq!(lb_configs[0].name, "pick_first"); + assert_eq!(lb_configs[0].config, json!({ "shuffleAddressList": true })); + assert_eq!(lb_configs[1].name, "round_robin"); + assert_eq!(lb_configs[1].config, json!({})); + + // Verify Method Config. + let method_configs = sc.method_config.unwrap(); + assert_eq!(method_configs.len(), 1); + let mc = &method_configs[0]; + assert_eq!(mc.name.len(), 2); + assert_eq!(mc.name[0].service, "grpc.examples.echo.Echo"); + assert_eq!(mc.name[0].method, Some("Echo".to_string())); + assert_eq!(mc.name[1].service, "grpc.examples.echo.Echo2"); + assert_eq!(mc.name[1].method, None); + + assert_eq!(mc.timeout, Some(Duration::new(1, 500_000_000))); + assert_eq!(mc.max_request_message_bytes, Some(1024)); + assert_eq!(mc.max_response_message_bytes, Some(2048)); + + // Verify Retry Policy. + let rp = mc.retry_policy.as_ref().unwrap(); + assert_eq!(rp.max_attempts, 3); + assert_eq!(rp.initial_backoff, Duration::new(0, 100_000_000)); + assert_eq!(rp.max_backoff, Duration::from_secs(1)); + assert_eq!(rp.backoff_multiplier, 2.0); + assert_eq!(rp.retryable_status_codes, vec!["UNAVAILABLE", "INTERNAL"]); + + // Verify Retry Throttling. + let rt = sc.retry_throttling.unwrap(); + assert_eq!(rt.max_tokens, 100); + assert_eq!(rt.token_ratio, 0.1); + } + + #[test] + fn test_invalid_service_config_parsing() { + // Bad JSON formatting. + assert!(ServiceConfig::parse("{").is_err()); + + // Invalid max attempts. + let json_data = json!({ + "methodConfig": [{ + "name": [{ "service": "foo" }], + "retryPolicy": { + "maxAttempts": 1, // Invalid, must be > 1. + "initialBackoff": "1s", + "maxBackoff": "1s", + "backoffMultiplier": 2.0, + "retryableStatusCodes": [] + } + }] + }); + assert!(ServiceConfig::parse(&json_data.to_string()).is_err()); + + // Invalid max tokens. + let json_data = json!({ + "retryThrottling": { + "maxTokens": 1001, // Invalid, max is 1000. + "tokenRatio": 0.1 + } + }); + assert!(ServiceConfig::parse(&json_data.to_string()).is_err()); + + // Empty method name service. + let json_data = json!({ + "methodConfig": [{ + "name": [{ "service": "" }] + }] + }); + assert!(ServiceConfig::parse(&json_data.to_string()).is_err()); + } +} diff --git a/grpc/src/inmemory/mod.rs b/grpc/src/inmemory/mod.rs index 4bba7357f..0a94a5a2e 100644 --- a/grpc/src/inmemory/mod.rs +++ b/grpc/src/inmemory/mod.rs @@ -428,9 +428,14 @@ impl Resolver for InMemoryResolver { let _ = channel_controller.update(ResolverUpdate { endpoints: Ok(endpoints), service_config: Ok(Some(ServiceConfig { - load_balancing_policy: Some( - crate::client::service_config::LbPolicyType::RoundRobin, - ), + load_balancing_config: Some(vec![ + crate::client::service_config::LoadBalancingConfig { + name: "round_robin".to_string(), + config: serde_json::Value::Object(serde_json::Map::new()), + } + ]), + method_config: None, + retry_throttling: None, })), ..Default::default() }); From 52589c3f20ac5eb01d58f52b9b1941f8ec8658f0 Mon Sep 17 00:00:00 2001 From: Nathaniel Ford Date: Mon, 13 Jul 2026 20:31:40 +0000 Subject: [PATCH 2/6] Streamlining deserialization --- grpc/src/client/service_config/json.rs | 70 +++++--------------------- grpc/src/inmemory/mod.rs | 2 +- 2 files changed, 13 insertions(+), 59 deletions(-) diff --git a/grpc/src/client/service_config/json.rs b/grpc/src/client/service_config/json.rs index 9771fb494..a5cc50940 100644 --- a/grpc/src/client/service_config/json.rs +++ b/grpc/src/client/service_config/json.rs @@ -25,6 +25,10 @@ use serde::Deserialize; use std::time::Duration; +// See [service config format notes](https://protobuf.dev/reference/protobuf/google.protobuf/#duration) +// for more on duration parsing. The format is a string with a number followed +// by an optional fraction and the letter 's' for seconds. For example, "1s", +// "1.5s", "0.000000001s" are valid durations. pub(crate) fn parse_duration(s: &str) -> Result { if !s.ends_with('s') { return Err("duration string must end with 's'".to_string()); @@ -65,72 +69,22 @@ where parse_duration(&s).map_err(serde::de::Error::custom) } +// Required to ensure that an optional Duraction can be parsed even when +// missing. pub(crate) fn deserialize_duration_opt<'de, D>( deserializer: D, ) -> Result, D::Error> where D: serde::Deserializer<'de>, { - struct OptVisitor; - impl<'de> serde::de::Visitor<'de> for OptVisitor { - type Value = Option; - - fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str("a string representing a duration or null") - } - - fn visit_none(self) -> Result - where - E: serde::de::Error, - { - Ok(None) - } - - fn visit_some(self, deserializer: D) -> Result - where - D: serde::Deserializer<'de>, - { - let s = String::deserialize(deserializer)?; - parse_duration(&s) - .map(Some) - .map_err(serde::de::Error::custom) - } - } - deserializer.deserialize_option(OptVisitor) -} - -pub(crate) fn deserialize_int64<'de, D>(deserializer: D) -> Result -where - D: serde::Deserializer<'de>, -{ - struct Visitor; - impl<'de> serde::de::Visitor<'de> for Visitor { - type Value = i64; - fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str("an i64 or a string representing an i64") - } - fn visit_i64(self, v: i64) -> Result - where - E: serde::de::Error, - { - Ok(v) - } - fn visit_u64(self, v: u64) -> Result - where - E: serde::de::Error, - { - v.try_into().map_err(serde::de::Error::custom) - } - fn visit_str(self, v: &str) -> Result - where - E: serde::de::Error, - { - v.parse().map_err(serde::de::Error::custom) - } - } - deserializer.deserialize_any(Visitor) + let opt = Option::::deserialize(deserializer)?; + opt.map(|s| parse_duration(&s).map_err(serde::de::Error::custom)) + .transpose() } +// This included to allow deserializing both u32 and string (or "stringified") +// representations of u32, as well as null values. This is used for the +// `max_concurrent_streams` field in the service config. pub(crate) fn deserialize_uint32_opt<'de, D>(deserializer: D) -> Result, D::Error> where D: serde::Deserializer<'de>, diff --git a/grpc/src/inmemory/mod.rs b/grpc/src/inmemory/mod.rs index 0a94a5a2e..3f5d02517 100644 --- a/grpc/src/inmemory/mod.rs +++ b/grpc/src/inmemory/mod.rs @@ -432,7 +432,7 @@ impl Resolver for InMemoryResolver { crate::client::service_config::LoadBalancingConfig { name: "round_robin".to_string(), config: serde_json::Value::Object(serde_json::Map::new()), - } + }, ]), method_config: None, retry_throttling: None, From 6f653a7c4cec47814666a6d988171cda7584cd7d Mon Sep 17 00:00:00 2001 From: Nathaniel Ford Date: Tue, 21 Jul 2026 21:06:23 +0000 Subject: [PATCH 3/6] PR feedback fixes --- grpc/src/client/channel.rs | 42 +-- grpc/src/client/name_resolution/mod.rs | 37 +-- .../client/name_resolution/proxy_resolver.rs | 4 +- grpc/src/client/name_resolution/test_utils.rs | 6 +- grpc/src/client/service_config/json.rs | 41 +-- grpc/src/client/service_config/mod.rs | 309 ++++++++++++------ grpc/src/client/service_config/serde.rs | 249 ++++++++++++++ grpc/src/inmemory/mod.rs | 17 +- 8 files changed, 541 insertions(+), 164 deletions(-) create mode 100644 grpc/src/client/service_config/serde.rs diff --git a/grpc/src/client/channel.rs b/grpc/src/client/channel.rs index 4caa1db75..444f7e7b8 100644 --- a/grpc/src/client/channel.rs +++ b/grpc/src/client/channel.rs @@ -2,23 +2,23 @@ * * Copyright 2025 gRPC authors. * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to - * deal in the Software without restriction, including without limitation the - * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - * sell copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - * IN THE SOFTWARE. + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. * */ @@ -68,7 +68,7 @@ use crate::client::name_resolution::Target; use crate::client::name_resolution::dns; use crate::client::name_resolution::global_registry; use crate::client::name_resolution::{self}; -use crate::client::service_config::ServiceConfig; +use crate::client::service_config::ParseResult; use crate::client::stream_util::FailingRecvStream; use crate::client::subchannel::InternalSubchannel; use crate::client::subchannel::NopBackoff; @@ -268,8 +268,9 @@ impl PersistentChannel { } } - /// Returns the current state of the channel. If there is no underlying active channel, - /// returns Idle. If `connect` is true, will create a new active channel iff none exists. + /// Returns the current state of the channel. If there is no underlying + /// active channel, returns Idle. If `connect` is true, will create a + /// new active channel iff none exists. fn get_state(&self, connect: bool) -> ConnectivityState { // Done this away to avoid potentially locking twice. let active_channel = if connect { @@ -286,8 +287,9 @@ impl PersistentChannel { active_channel.lb_watcher.cur().connectivity_state } - /// Gets the underlying active channel. If there is no current connection, it will create one. - /// This cannot fail and will always return a valid active channel. + /// Gets the underlying active channel. If there is no current connection, + /// it will create one. This cannot fail and will always return a valid + /// active channel. fn get_active_channel(&self) -> Arc { let mut active_channel = self.active_channel.lock().unwrap(); @@ -301,8 +303,10 @@ impl PersistentChannel { // A channel that is not idle (connecting, ready, or erroring). struct ActiveChannel { - abort_handle: Box, // Work scheduler task killed when ActiveChannel is dropped. - lb_watcher: Arc>, // For getting the channel connectivity state and pickers for RPCs. + abort_handle: Box, /* Work scheduler task killed when ActiveChannel is + * dropped. */ + lb_watcher: Arc>, /* For getting the channel connectivity state and pickers + * for RPCs. */ } impl ActiveChannel { @@ -481,8 +485,8 @@ impl name_resolution::ChannelController for ResolverChannelController { .map_err(|err| err.to_string()) } - fn parse_service_config(&self, config: &str) -> Result { - Err("service configs not supported".to_string()) + fn parse_service_config(&self, config: &str) -> ParseResult { + ParseResult(Err("service configs not supported".to_string())) } } diff --git a/grpc/src/client/name_resolution/mod.rs b/grpc/src/client/name_resolution/mod.rs index f7517a3e3..35eaf7371 100644 --- a/grpc/src/client/name_resolution/mod.rs +++ b/grpc/src/client/name_resolution/mod.rs @@ -2,23 +2,23 @@ * * Copyright 2025 gRPC authors. * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to - * deal in the Software without restriction, including without limitation the - * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - * sell copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - * IN THE SOFTWARE. + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. * */ @@ -34,14 +34,15 @@ use std::hash::Hash; use std::str::FromStr; use std::sync::Arc; -use percent_encoding::AsciiSet; -use percent_encoding::NON_ALPHANUMERIC; use percent_encoding::percent_decode_str; use percent_encoding::utf8_percent_encode; +use percent_encoding::AsciiSet; +use percent_encoding::NON_ALPHANUMERIC; use url::Url; use crate::attributes::Attributes; use crate::byte_str::ByteStr; +use crate::client::service_config::ParseResult; use crate::client::service_config::ServiceConfig; use crate::rt::GrpcRuntime; @@ -265,7 +266,7 @@ pub trait ChannelController: Send + Sync { /// Parses the provided JSON service config and returns an instance of a /// ParsedServiceConfig. - fn parse_service_config(&self, config: &str) -> Result; + fn parse_service_config(&self, config: &str) -> ParseResult; } #[derive(Clone, Debug)] @@ -409,8 +410,8 @@ impl NopResolver { #[cfg(test)] mod test { - use std::collections::HashMap; use std::collections::hash_map::DefaultHasher; + use std::collections::HashMap; use std::hash::Hash; use std::hash::Hasher; @@ -492,11 +493,9 @@ mod test { let input = "dns:///foo%FFbar"; let target: Result = input.parse(); assert!(target.is_err()); - assert!( - target - .unwrap_err() - .contains("invalid UTF-8 character in target path") - ); + assert!(target + .unwrap_err() + .contains("invalid UTF-8 character in target path")); } // This test ensures that the Address struct correctly maintains its diff --git a/grpc/src/client/name_resolution/proxy_resolver.rs b/grpc/src/client/name_resolution/proxy_resolver.rs index 6e6488794..a58eed074 100644 --- a/grpc/src/client/name_resolution/proxy_resolver.rs +++ b/grpc/src/client/name_resolution/proxy_resolver.rs @@ -36,7 +36,7 @@ use crate::client::name_resolution::ResolverOptions; use crate::client::name_resolution::ResolverUpdate; use crate::client::name_resolution::Target; use crate::client::name_resolution::dns; -use crate::client::service_config::ServiceConfig; +use crate::client::service_config::ParseResult; use crate::client::transport::ProxyOptions; use crate::credentials::common::Authority; @@ -227,7 +227,7 @@ impl<'a> ChannelController for InterceptingController<'a> { self.inner.update(update) } - fn parse_service_config(&self, config: &str) -> Result { + fn parse_service_config(&self, config: &str) -> ParseResult { self.inner.parse_service_config(config) } } diff --git a/grpc/src/client/name_resolution/test_utils.rs b/grpc/src/client/name_resolution/test_utils.rs index 75db9f9f9..913c776fe 100644 --- a/grpc/src/client/name_resolution/test_utils.rs +++ b/grpc/src/client/name_resolution/test_utils.rs @@ -29,7 +29,7 @@ use tokio::sync::mpsc; use crate::client::name_resolution::ChannelController; use crate::client::name_resolution::ResolverUpdate; use crate::client::name_resolution::WorkScheduler; -use crate::client::service_config::ServiceConfig; +use crate::client::service_config::ParseResult; /// A work scheduler for testing. pub(crate) struct TestWorkScheduler { @@ -80,7 +80,7 @@ impl ChannelController for TestChannelController { self.update_result.clone() } - fn parse_service_config(&self, _: &str) -> Result { - Err("Unimplemented".to_string()) + fn parse_service_config(&self, _: &str) -> ParseResult { + ParseResult(Err("Unimplemented".to_string())) } } diff --git a/grpc/src/client/service_config/json.rs b/grpc/src/client/service_config/json.rs index a5cc50940..f659f1d8a 100644 --- a/grpc/src/client/service_config/json.rs +++ b/grpc/src/client/service_config/json.rs @@ -2,33 +2,34 @@ * * Copyright 2026 gRPC authors. * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to - * deal in the Software without restriction, including without limitation the - * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - * sell copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - * IN THE SOFTWARE. + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. * */ -use serde::Deserialize; use std::time::Duration; -// See [service config format notes](https://protobuf.dev/reference/protobuf/google.protobuf/#duration) -// for more on duration parsing. The format is a string with a number followed -// by an optional fraction and the letter 's' for seconds. For example, "1s", -// "1.5s", "0.000000001s" are valid durations. +use serde::Deserialize; + +/// Service Config uses the [protobuf Duration format](https://protobuf.dev/reference/protobuf/google.protobuf/#duration) +/// when parsing. The format is a string with a number followed +/// by an optional fraction and the letter 's' for seconds. For example, "1s", +/// "1.5s", "0.000000001s" are valid durations. pub(crate) fn parse_duration(s: &str) -> Result { if !s.ends_with('s') { return Err("duration string must end with 's'".to_string()); @@ -82,9 +83,9 @@ where .transpose() } -// This included to allow deserializing both u32 and string (or "stringified") -// representations of u32, as well as null values. This is used for the -// `max_concurrent_streams` field in the service config. +// This is included to allow deserializing both u32 and string (or +// "stringified") representations of u32, as well as null values. This is used, +// for example, for the `max_request_message_bytes` field in the service config. pub(crate) fn deserialize_uint32_opt<'de, D>(deserializer: D) -> Result, D::Error> where D: serde::Deserializer<'de>, @@ -130,9 +131,10 @@ where #[cfg(test)] mod test { - use super::*; use serde_json::json; + use super::*; + #[test] fn test_parse_duration() { assert_eq!(parse_duration("1s").unwrap(), Duration::from_secs(1)); @@ -153,6 +155,7 @@ mod test { assert!(parse_duration("1.s").is_err()); assert!(parse_duration("1.0000000001s").is_err()); assert!(parse_duration("as").is_err()); + assert!(parse_duration(".5s").is_err()); } #[test] diff --git a/grpc/src/client/service_config/mod.rs b/grpc/src/client/service_config/mod.rs index a7c52fc7f..aa139698a 100644 --- a/grpc/src/client/service_config/mod.rs +++ b/grpc/src/client/service_config/mod.rs @@ -2,142 +2,177 @@ * * Copyright 2026 gRPC authors. * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to - * deal in the Software without restriction, including without limitation the - * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - * sell copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - * IN THE SOFTWARE. + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. * */ - -use serde::{Deserialize, Serialize}; use std::time::Duration; pub(crate) mod json; +pub(crate) mod serde; -/// An in-memory representation of a service config, provided to gRPC as a JSON object. -#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] -#[serde(rename_all = "camelCase")] +/// An in-memory representation of a service config, provided to gRPC as a JSON +/// object. +#[derive(Debug, Clone, PartialEq)] pub struct ServiceConfig { + pub(crate) load_balancing_policy: Option, /// Ordered list of LB policies. The first supported one is used. - pub load_balancing_config: Option>, + pub(crate) load_balancing_config: Option>, /// Per-method configuration overrides. - pub method_config: Option>, + pub(crate) method_config: Option>, /// Global retry throttling parameters. - pub retry_throttling: Option, + pub(crate) retry_throttling: Option, + /// Health check configuration. + pub(crate) health_check_config: Option, + /// Connection scaling configuration. + pub(crate) connection_scaling: Option, } -#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] -#[serde(rename_all = "camelCase")] +#[derive(Debug, Clone, PartialEq)] pub struct MethodConfig { /// List of methods this config applies to. - pub name: Vec, - #[serde(default, deserialize_with = "json::deserialize_duration_opt")] - pub timeout: Option, - pub retry_policy: Option, - #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] + pub(crate) name: Vec, + pub(crate) wait_for_ready: Option, + pub(crate) timeout: Option, + pub(crate) retry_policy: Option, + pub(crate) hedging_policy: Option, pub max_request_message_bytes: Option, - #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] - pub max_response_message_bytes: Option, + pub(crate) max_response_message_bytes: Option, } -#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct MethodName { - /// e.g., "grpc.examples.echo.Echo" or "grpc.examples.echo.Echo/Echo". /// The service name (fully qualified). - pub service: String, + /// e.g., for "grpc.examples.echo.Echo/Echo" this would be + /// "grpc.examples.echo.Echo". + pub(crate) service: String, /// If None, applies to all methods in the service. - pub method: Option, + /// e.g., for "grpc.examples.echo.Echo/Echo" this would be "Echo". + pub(crate) method: Option, } -#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] -#[serde(rename_all = "camelCase")] +#[derive(Debug, Clone, PartialEq)] pub struct RetryThrottlingPolicy { - pub max_tokens: u32, - pub token_ratio: f32, + pub(crate) max_tokens: u32, + pub(crate) token_ratio: f32, } -#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] -#[serde(rename_all = "camelCase")] +#[derive(Debug, Clone, PartialEq)] pub struct RetryPolicy { - pub max_attempts: u32, - #[serde(deserialize_with = "json::deserialize_duration")] - pub initial_backoff: Duration, - #[serde(deserialize_with = "json::deserialize_duration")] - pub max_backoff: Duration, - pub backoff_multiplier: f32, - pub retryable_status_codes: Vec, + pub(crate) max_attempts: u32, + pub(crate) initial_backoff: Duration, + pub(crate) max_backoff: Duration, + pub(crate) backoff_multiplier: f32, + pub(crate) retryable_status_codes: Vec, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct HedgingPolicy { + pub(crate) max_attempts: u32, + pub(crate) hedging_delay: Duration, + pub(crate) non_fatal_status_codes: Option>, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct HealthCheckConfig { + pub(crate) service_name: Option, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct ConnectionScaling { + pub(crate) max_connections_per_subchannel: u32, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct LoadBalancingConfig { - pub name: String, - pub config: serde_json::Value, + pub(crate) name: String, + pub(crate) config: serde_json::Value, } -impl<'de> Deserialize<'de> for LoadBalancingConfig { - fn deserialize(deserializer: D) -> Result - where - D: serde::Deserializer<'de>, - { - struct Visitor; - impl<'de> serde::de::Visitor<'de> for Visitor { - type Value = LoadBalancingConfig; - fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str( - "a map with a single key-value pair representing a load balancing policy", - ) - } - fn visit_map(self, mut map: A) -> Result - where - A: serde::de::MapAccess<'de>, - { - if let Some(name) = map.next_key::()? { - let config = map.next_value::()?; - if map.next_key::()?.is_some() { - return Err(serde::de::Error::custom("map has more than one key")); - } - Ok(LoadBalancingConfig { name, config }) - } else { - Err(serde::de::Error::custom("map is empty")) - } - } - } - deserializer.deserialize_map(Visitor) +#[derive(Debug, Clone, PartialEq)] +pub enum RetryOrHedgingPolicy { + Retry(RetryPolicy), + Hedging(HedgingPolicy), +} + +impl MethodConfig { + pub fn retry_or_hedging_policy(&self) -> Option { + self.retry_policy + .as_ref() + .map(|rp| RetryOrHedgingPolicy::Retry(rp.clone())) + .or_else(|| { + self.hedging_policy + .as_ref() + .map(|hp| RetryOrHedgingPolicy::Hedging(hp.clone())) + }) } } -impl Serialize for LoadBalancingConfig { - fn serialize(&self, serializer: S) -> Result - where - S: serde::Serializer, - { - use serde::ser::SerializeMap; - let mut map = serializer.serialize_map(Some(1))?; - map.serialize_entry(&self.name, &self.config)?; - map.end() +#[derive(Debug, Clone, PartialEq)] +pub struct ParseResult(pub Result); + +impl ParseResult { + pub fn is_ok(&self) -> bool { + self.0.is_ok() + } + + pub fn is_err(&self) -> bool { + self.0.is_err() + } + + pub fn unwrap(self) -> ServiceConfig { + self.0.unwrap() + } + + pub fn unwrap_err(self) -> String { + self.0.unwrap_err() + } + + pub fn into_inner(self) -> Result { + self.0 + } +} + +impl std::ops::Deref for ParseResult { + type Target = Result; + + fn deref(&self) -> &Self::Target { + &self.0 } } impl ServiceConfig { /// Parses a service configuration from a JSON string. - pub fn parse(config_json: &str) -> Result { - let config: Self = serde_json::from_str(config_json) - .map_err(|e| format!("failed to deserialize service config JSON: {}", e))?; - config.validate()?; - Ok(config) + pub fn parse(config_json: &str) -> ParseResult { + let config_serde: serde::ServiceConfigSerDe = match serde_json::from_str(config_json) { + Ok(c) => c, + Err(e) => { + return ParseResult(Err(format!( + "failed to deserialize service config JSON: {}", + e + ))); + } + }; + let config: Self = config_serde.into(); + if let Err(e) = config.validate() { + return ParseResult(Err(e)); + } + ParseResult(Ok(config)) } fn validate(&self) -> Result<(), String> { @@ -154,6 +189,12 @@ impl ServiceConfig { )); } } + if mc.retry_policy.is_some() && mc.hedging_policy.is_some() { + return Err(format!( + "method_config[{}] cannot have both retryPolicy and hedgingPolicy defined", + i + )); + } if let Some(ref rp) = mc.retry_policy { if rp.max_attempts <= 1 { return Err(format!( @@ -180,6 +221,14 @@ impl ServiceConfig { )); } } + if let Some(ref hp) = mc.hedging_policy + && hp.max_attempts <= 1 + { + return Err(format!( + "method_config[{}].hedging_policy.max_attempts must be > 1", + i + )); + } } } @@ -198,9 +247,10 @@ impl ServiceConfig { #[cfg(test)] mod test { - use super::*; use serde_json::json; + use super::*; + #[test] fn test_valid_service_config_parsing() { let json_data = json!({ @@ -224,11 +274,28 @@ mod test { }, "maxRequestMessageBytes": 1024, "maxResponseMessageBytes": "2048" + }, + { + "name": [ + { "service": "grpc.examples.echo.EchoHedging" } + ], + "waitForReady": true, + "hedgingPolicy": { + "maxAttempts": 3, + "hedgingDelay": "0.5s", + "nonFatalStatusCodes": ["UNAVAILABLE"] + } } ], "retryThrottling": { "maxTokens": 100, "tokenRatio": 0.1 + }, + "healthCheckConfig": { + "serviceName": "grpc.health.v1.Health" + }, + "connectionScaling": { + "maxConnectionsPerSubchannel": 20 } }); @@ -244,7 +311,7 @@ mod test { // Verify Method Config. let method_configs = sc.method_config.unwrap(); - assert_eq!(method_configs.len(), 1); + assert_eq!(method_configs.len(), 2); let mc = &method_configs[0]; assert_eq!(mc.name.len(), 2); assert_eq!(mc.name[0].service, "grpc.examples.echo.Echo"); @@ -264,10 +331,36 @@ mod test { assert_eq!(rp.backoff_multiplier, 2.0); assert_eq!(rp.retryable_status_codes, vec!["UNAVAILABLE", "INTERNAL"]); + let mc2 = &method_configs[1]; + assert_eq!(mc2.wait_for_ready, Some(true)); + let hp = mc2.hedging_policy.as_ref().unwrap(); + assert_eq!(hp.max_attempts, 3); + assert_eq!(hp.hedging_delay, Duration::from_millis(500)); + assert_eq!( + hp.non_fatal_status_codes, + Some(vec!["UNAVAILABLE".to_string()]) + ); + assert_eq!( + mc2.retry_or_hedging_policy(), + Some(RetryOrHedgingPolicy::Hedging(hp.clone())) + ); + // Verify Retry Throttling. let rt = sc.retry_throttling.unwrap(); assert_eq!(rt.max_tokens, 100); assert_eq!(rt.token_ratio, 0.1); + + assert_eq!( + sc.health_check_config.as_ref().unwrap().service_name, + Some("grpc.health.v1.Health".to_string()) + ); + assert_eq!( + sc.connection_scaling + .as_ref() + .unwrap() + .max_connections_per_subchannel, + 20 + ); } #[test] @@ -307,4 +400,38 @@ mod test { }); assert!(ServiceConfig::parse(&json_data.to_string()).is_err()); } + + #[test] + fn test_legacy_lb_policy_fallback() { + let json_data = json!({ + "loadBalancingPolicy": "round_robin" + }); + let sc = ServiceConfig::parse(&json_data.to_string()).unwrap(); + assert_eq!(sc.load_balancing_policy, Some("round_robin".to_string())); + let lb_configs = sc.load_balancing_config.unwrap(); + assert_eq!(lb_configs.len(), 1); + assert_eq!(lb_configs[0].name, "round_robin"); + assert_eq!(lb_configs[0].config, json!({})); + } + + #[test] + fn test_retry_hedging_mutual_exclusivity() { + let json_data = json!({ + "methodConfig": [{ + "name": [{ "service": "foo" }], + "retryPolicy": { + "maxAttempts": 2, + "initialBackoff": "1s", + "maxBackoff": "1s", + "backoffMultiplier": 2.0, + "retryableStatusCodes": [] + }, + "hedgingPolicy": { + "maxAttempts": 2, + "hedgingDelay": "1s" + } + }] + }); + assert!(ServiceConfig::parse(&json_data.to_string()).is_err()); + } } diff --git a/grpc/src/client/service_config/serde.rs b/grpc/src/client/service_config/serde.rs new file mode 100644 index 000000000..f1afe3400 --- /dev/null +++ b/grpc/src/client/service_config/serde.rs @@ -0,0 +1,249 @@ +use std::time::Duration; + +use serde::Deserialize; +use serde::Serialize; + +use super::ConnectionScaling; +use super::HealthCheckConfig; +use super::HedgingPolicy; +use super::LoadBalancingConfig; +use super::MethodConfig; +use super::MethodName; +use super::RetryPolicy; +use super::RetryThrottlingPolicy; +use super::ServiceConfig; +use super::json; + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ServiceConfigSerDe { + pub(crate) load_balancing_policy: Option, + pub(crate) load_balancing_config: Option>, + pub(crate) method_config: Option>, + pub(crate) retry_throttling: Option, + pub(crate) health_check_config: Option, + pub(crate) connection_scaling: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct MethodConfigSerDe { + pub(crate) name: Vec, + pub(crate) wait_for_ready: Option, + #[serde(default, deserialize_with = "json::deserialize_duration_opt")] + pub(crate) timeout: Option, + pub(crate) retry_policy: Option, + pub(crate) hedging_policy: Option, + #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] + pub(crate) max_request_message_bytes: Option, + #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] + pub(crate) max_response_message_bytes: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +pub(crate) struct MethodNameSerDe { + pub(crate) service: String, + pub(crate) method: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RetryThrottlingPolicySerDe { + pub(crate) max_tokens: u32, + pub(crate) token_ratio: f32, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RetryPolicySerDe { + pub(crate) max_attempts: u32, + #[serde(deserialize_with = "json::deserialize_duration")] + pub(crate) initial_backoff: Duration, + #[serde(deserialize_with = "json::deserialize_duration")] + pub(crate) max_backoff: Duration, + pub(crate) backoff_multiplier: f32, + pub(crate) retryable_status_codes: Vec, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct HedgingPolicySerDe { + pub(crate) max_attempts: u32, + #[serde(deserialize_with = "json::deserialize_duration")] + pub(crate) hedging_delay: Duration, + #[serde(default)] + pub(crate) non_fatal_status_codes: Option>, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct HealthCheckConfigSerDe { + #[serde(rename = "serviceName", alias = "ServiceName")] + pub(crate) service_name: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ConnectionScalingSerDe { + #[serde(default = "default_max_connections_per_subchannel")] + pub(crate) max_connections_per_subchannel: u32, +} + +fn default_max_connections_per_subchannel() -> u32 { + 10 +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct LoadBalancingConfigSerDe { + pub(crate) name: String, + pub(crate) config: serde_json::Value, +} + +impl<'de> Deserialize<'de> for LoadBalancingConfigSerDe { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + struct Visitor; + impl<'de> serde::de::Visitor<'de> for Visitor { + type Value = LoadBalancingConfigSerDe; + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str( + "a map with a single key-value pair representing a load balancing policy", + ) + } + fn visit_map(self, mut map: A) -> Result + where + A: serde::de::MapAccess<'de>, + { + if let Some(name) = map.next_key::()? { + let config = map.next_value::()?; + if map.next_key::()?.is_some() { + return Err(serde::de::Error::custom("map has more than one key")); + } + Ok(LoadBalancingConfigSerDe { name, config }) + } else { + Err(serde::de::Error::custom("map is empty")) + } + } + } + deserializer.deserialize_map(Visitor) + } +} + +impl Serialize for LoadBalancingConfigSerDe { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + use serde::ser::SerializeMap; + let mut map = serializer.serialize_map(Some(1))?; + map.serialize_entry(&self.name, &self.config)?; + map.end() + } +} + +impl From for ServiceConfig { + fn from(dto: ServiceConfigSerDe) -> Self { + let mut lb_config = dto + .load_balancing_config + .map(|v| v.into_iter().map(Into::into).collect()); + if lb_config.is_none() + && let Some(ref lb_policy) = dto.load_balancing_policy + { + lb_config = Some(vec![LoadBalancingConfig { + name: lb_policy.clone(), + config: serde_json::Value::Object(serde_json::Map::new()), + }]); + } + Self { + load_balancing_policy: dto.load_balancing_policy, + load_balancing_config: lb_config, + method_config: dto + .method_config + .map(|v| v.into_iter().map(Into::into).collect()), + retry_throttling: dto.retry_throttling.map(Into::into), + health_check_config: dto.health_check_config.map(Into::into), + connection_scaling: dto.connection_scaling.map(Into::into), + } + } +} + +impl From for MethodConfig { + fn from(dto: MethodConfigSerDe) -> Self { + Self { + name: dto.name.into_iter().map(Into::into).collect(), + wait_for_ready: dto.wait_for_ready, + timeout: dto.timeout, + retry_policy: dto.retry_policy.map(Into::into), + hedging_policy: dto.hedging_policy.map(Into::into), + max_request_message_bytes: dto.max_request_message_bytes, + max_response_message_bytes: dto.max_response_message_bytes, + } + } +} + +impl From for MethodName { + fn from(dto: MethodNameSerDe) -> Self { + Self { + service: dto.service, + method: dto.method, + } + } +} + +impl From for RetryThrottlingPolicy { + fn from(dto: RetryThrottlingPolicySerDe) -> Self { + Self { + max_tokens: dto.max_tokens, + token_ratio: dto.token_ratio, + } + } +} + +impl From for RetryPolicy { + fn from(dto: RetryPolicySerDe) -> Self { + Self { + max_attempts: dto.max_attempts, + initial_backoff: dto.initial_backoff, + max_backoff: dto.max_backoff, + backoff_multiplier: dto.backoff_multiplier, + retryable_status_codes: dto.retryable_status_codes, + } + } +} + +impl From for HedgingPolicy { + fn from(dto: HedgingPolicySerDe) -> Self { + Self { + max_attempts: dto.max_attempts, + hedging_delay: dto.hedging_delay, + non_fatal_status_codes: dto.non_fatal_status_codes, + } + } +} + +impl From for HealthCheckConfig { + fn from(dto: HealthCheckConfigSerDe) -> Self { + Self { + service_name: dto.service_name, + } + } +} + +impl From for ConnectionScaling { + fn from(dto: ConnectionScalingSerDe) -> Self { + Self { + max_connections_per_subchannel: dto.max_connections_per_subchannel, + } + } +} + +impl From for LoadBalancingConfig { + fn from(dto: LoadBalancingConfigSerDe) -> Self { + Self { + name: dto.name, + config: dto.config, + } + } +} diff --git a/grpc/src/inmemory/mod.rs b/grpc/src/inmemory/mod.rs index 3f5d02517..38a86052a 100644 --- a/grpc/src/inmemory/mod.rs +++ b/grpc/src/inmemory/mod.rs @@ -55,7 +55,6 @@ use crate::client::name_resolution::ResolverOptions; use crate::client::name_resolution::ResolverUpdate; use crate::client::name_resolution::Target; use crate::client::name_resolution::global_registry as global_resolver_registry; -use crate::client::service_config::ServiceConfig; use crate::client::transport::GLOBAL_TRANSPORT_REGISTRY; use crate::client::transport::SecurityOpts; use crate::client::transport::Transport; @@ -425,18 +424,14 @@ impl Resolver for InMemoryResolver { }) .collect(); + let service_config = channel_controller + .parse_service_config(r#"{"loadBalancingConfig": [{"round_robin": {}}]}"#) + .into_inner() + .map(Some); + let _ = channel_controller.update(ResolverUpdate { endpoints: Ok(endpoints), - service_config: Ok(Some(ServiceConfig { - load_balancing_config: Some(vec![ - crate::client::service_config::LoadBalancingConfig { - name: "round_robin".to_string(), - config: serde_json::Value::Object(serde_json::Map::new()), - }, - ]), - method_config: None, - retry_throttling: None, - })), + service_config, ..Default::default() }); } From b606dff09f87c3684597582875ce5078dc87d35e Mon Sep 17 00:00:00 2001 From: Nathaniel Ford Date: Tue, 21 Jul 2026 21:13:09 +0000 Subject: [PATCH 4/6] Cargo fmt fix --- grpc/src/client/name_resolution/mod.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/grpc/src/client/name_resolution/mod.rs b/grpc/src/client/name_resolution/mod.rs index 35eaf7371..d0f452317 100644 --- a/grpc/src/client/name_resolution/mod.rs +++ b/grpc/src/client/name_resolution/mod.rs @@ -34,10 +34,10 @@ use std::hash::Hash; use std::str::FromStr; use std::sync::Arc; -use percent_encoding::percent_decode_str; -use percent_encoding::utf8_percent_encode; use percent_encoding::AsciiSet; use percent_encoding::NON_ALPHANUMERIC; +use percent_encoding::percent_decode_str; +use percent_encoding::utf8_percent_encode; use url::Url; use crate::attributes::Attributes; @@ -410,8 +410,8 @@ impl NopResolver { #[cfg(test)] mod test { - use std::collections::hash_map::DefaultHasher; use std::collections::HashMap; + use std::collections::hash_map::DefaultHasher; use std::hash::Hash; use std::hash::Hasher; @@ -493,9 +493,11 @@ mod test { let input = "dns:///foo%FFbar"; let target: Result = input.parse(); assert!(target.is_err()); - assert!(target - .unwrap_err() - .contains("invalid UTF-8 character in target path")); + assert!( + target + .unwrap_err() + .contains("invalid UTF-8 character in target path") + ); } // This test ensures that the Address struct correctly maintains its From c8b636d7a16ccd75c5ce1d80c0a402fb440426e1 Mon Sep 17 00:00:00 2001 From: Nathaniel Ford Date: Wed, 22 Jul 2026 20:14:39 +0000 Subject: [PATCH 5/6] Add GrpcDuration, remove u32 custom deserialization, fix formatting --- grpc/src/client/channel.rs | 7 +- grpc/src/client/name_resolution/test_utils.rs | 2 +- grpc/src/client/service_config/duration.rs | 226 ++++++++++++++++++ grpc/src/client/service_config/json.rs | 223 ----------------- grpc/src/client/service_config/mod.rs | 93 +++---- grpc/src/client/service_config/serde.rs | 42 +++- grpc/src/inmemory/mod.rs | 4 +- 7 files changed, 296 insertions(+), 301 deletions(-) create mode 100644 grpc/src/client/service_config/duration.rs delete mode 100644 grpc/src/client/service_config/json.rs diff --git a/grpc/src/client/channel.rs b/grpc/src/client/channel.rs index 34f7ccfdd..65f81946e 100644 --- a/grpc/src/client/channel.rs +++ b/grpc/src/client/channel.rs @@ -249,8 +249,9 @@ struct PersistentChannel { } impl PersistentChannel { - /// Returns the current state of the channel. If there is no underlying active channel, - /// returns Idle. If `connect` is true, will create a new active channel iff none exists. + /// Returns the current state of the channel. If there is no underlying active + /// channel, returns Idle. If `connect` is true, will create a new active + /// channel iff none exists. fn get_state(&self, connect: bool) -> ConnectivityState { // Done this away to avoid potentially locking twice. let active_channel = if connect { @@ -469,7 +470,7 @@ impl name_resolution::ChannelController for ResolverChannelController { } fn parse_service_config(&self, config: &str) -> ParseResult { - ParseResult(Err("service configs not supported".to_string())) + Err("service configs not supported".to_string()) } } diff --git a/grpc/src/client/name_resolution/test_utils.rs b/grpc/src/client/name_resolution/test_utils.rs index 913c776fe..e2b116162 100644 --- a/grpc/src/client/name_resolution/test_utils.rs +++ b/grpc/src/client/name_resolution/test_utils.rs @@ -81,6 +81,6 @@ impl ChannelController for TestChannelController { } fn parse_service_config(&self, _: &str) -> ParseResult { - ParseResult(Err("Unimplemented".to_string())) + Err("Unimplemented".to_string()) } } diff --git a/grpc/src/client/service_config/duration.rs b/grpc/src/client/service_config/duration.rs new file mode 100644 index 000000000..6292849ed --- /dev/null +++ b/grpc/src/client/service_config/duration.rs @@ -0,0 +1,226 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +// TODO(nathanielford) Move this to the internal crate once we have one. + +use std::time::Duration; + +use serde::Deserialize; +use serde::Serialize; +use serde::Serializer; + +// Wraps std::time::Duration to provide custom serialization and deserialization +// for gRPC service config, according to the protobuf Duration format. +#[derive(PartialEq, Debug, Clone)] +pub(crate) struct GrpcDuration(pub(crate) Duration); + +impl From for Duration { + fn from(d: GrpcDuration) -> Self { + d.0 + } +} + +impl From for GrpcDuration { + fn from(d: Duration) -> Self { + GrpcDuration(d) + } +} + +impl std::ops::Deref for GrpcDuration { + type Target = Duration; + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +/// Service Config uses the [protobuf Duration format](https://protobuf.dev/reference/protobuf/google.protobuf/#duration) +/// when parsing. The format is a string with a number followed +/// by an optional fraction and the letter 's' for seconds. For example, "1s", +/// "1.5s", "0.000000001s" are valid durations. +impl<'de> Deserialize<'de> for GrpcDuration { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let raw_value = String::deserialize(deserializer)?; + let duration = parse_duration(&raw_value).map_err(serde::de::Error::custom)?; + Ok(GrpcDuration(duration)) + } +} + +impl Serialize for GrpcDuration { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&format_duration(&self.0)) + } +} + +// Parsing logic, isolated for testing and to reduce serde monomorphization +// footprint. +fn parse_duration(s: &str) -> Result { + if !s.ends_with('s') { + return Err("duration string must end with 's'".to_string()); + } + let s = &s[..s.len() - 1]; // strip 's'. + let mut parts = s.splitn(2, '.'); + let secs_str = parts + .next() + .ok_or_else(|| "empty duration string".to_string())?; + let secs: u64 = secs_str + .parse() + .map_err(|e| format!("failed to parse seconds: {}", e))?; + + let nanos = if let Some(fraction_str) = parts.next() { + if fraction_str.is_empty() { + return Err("empty fraction part".to_string()); + } + if fraction_str.len() > 9 { + return Err("fraction part has more than 9 digits".to_string()); + } + let fraction_val: u32 = fraction_str + .parse() + .map_err(|e| format!("failed to parse fraction: {}", e))?; + let pad = 9 - fraction_str.len(); + fraction_val * 10u32.pow(pad as u32) + } else { + 0 + }; + Ok(Duration::new(secs, nanos)) +} + +// This is the inverse of parse_duration. +fn format_duration(duration: &Duration) -> String { + let secs = duration.as_secs(); + let nanos = duration.subsec_nanos(); + + if nanos == 0 { + format!("{}s", secs) + } else { + let frac = format!("{:09}", nanos); + let trimmed_frac = frac.trim_end_matches('0'); + format!("{}.{}s", secs, trimmed_frac) + } +} + +#[cfg(test)] +mod test { + use serde_json::json; + + use super::*; + + #[test] + fn test_parse_duration() { + assert_eq!(parse_duration("1s").unwrap(), Duration::from_secs(1)); + assert_eq!(parse_duration("0s").unwrap(), Duration::from_secs(0)); + assert_eq!( + parse_duration("1.5s").unwrap(), + Duration::new(1, 500_000_000) + ); + assert_eq!(parse_duration("0.000000001s").unwrap(), Duration::new(0, 1)); + assert_eq!(parse_duration("1.0000001s").unwrap(), Duration::new(1, 100)); + assert_eq!( + parse_duration("1.0001s").unwrap(), + Duration::new(1, 100_000) + ); + + assert!(parse_duration("").is_err()); + assert!(parse_duration("1").is_err()); + assert!(parse_duration("1.s").is_err()); + assert!(parse_duration("1.0000000001s").is_err()); + assert!(parse_duration("as").is_err()); + assert!(parse_duration(".5s").is_err()); + } + + #[test] + fn test_serialize_duration() { + assert_eq!( + serde_json::to_value(GrpcDuration(Duration::from_secs(1))).unwrap(), + json!("1s") + ); + assert_eq!( + serde_json::to_value(GrpcDuration(Duration::from_secs(0))).unwrap(), + json!("0s") + ); + assert_eq!( + serde_json::to_value(GrpcDuration(Duration::new(1, 500_000_000))).unwrap(), + json!("1.5s") + ); + assert_eq!( + serde_json::to_value(GrpcDuration(Duration::new(0, 1))).unwrap(), + json!("0.000000001s") + ); + assert_eq!( + serde_json::to_value(GrpcDuration(Duration::new(1, 100))).unwrap(), + json!("1.0000001s") + ); + assert_eq!( + serde_json::to_value(GrpcDuration(Duration::new(1, 100_000))).unwrap(), + json!("1.0001s") + ); + } + + #[test] + fn test_parse_format_inverse() { + let canonical_strings = ["1s", "0s", "1.5s", "0.000000001s", "1.0000001s", "1.0001s"]; + for s in canonical_strings { + let parsed = parse_duration(s).unwrap(); + let formatted = format_duration(&parsed); + assert_eq!(formatted, s); + } + + let durations = [ + Duration::from_secs(1), + Duration::from_secs(0), + Duration::new(1, 500_000_000), + Duration::new(0, 1), + Duration::new(1, 100), + Duration::new(1, 100_000), + ]; + for d in durations { + let formatted = format_duration(&d); + let parsed = parse_duration(&formatted).unwrap(); + assert_eq!(parsed, d); + } + } + + #[test] + fn test_bijective_roundtrip() { + let test_cases = [ + ("1s", GrpcDuration(Duration::from_secs(1))), + ("0s", GrpcDuration(Duration::from_secs(0))), + ("1.5s", GrpcDuration(Duration::new(1, 500_000_000))), + ("0.000000001s", GrpcDuration(Duration::new(0, 1))), + ("1.0000001s", GrpcDuration(Duration::new(1, 100))), + ("1.0001s", GrpcDuration(Duration::new(1, 100_000))), + ]; + for (s, expected) in test_cases { + let val: GrpcDuration = serde_json::from_value(json!(s)).unwrap(); + assert_eq!(val, expected); // Deserialized value is correct. + let res = serde_json::to_value(&val).unwrap(); + assert_eq!(res, json!(s)); // Re-serialized value is equivalent. + } + } +} diff --git a/grpc/src/client/service_config/json.rs b/grpc/src/client/service_config/json.rs deleted file mode 100644 index f659f1d8a..000000000 --- a/grpc/src/client/service_config/json.rs +++ /dev/null @@ -1,223 +0,0 @@ -/* - * - * Copyright 2026 gRPC authors. - * - * Permission is hereby granted, free of charge, to any person obtaining a - * copy of this software and associated documentation files (the "Software"), - * to deal in the Software without restriction, including without limitation - * the rights to use, copy, modify, merge, publish, distribute, sublicense, - * and/or sell copies of the Software, and to permit persons to whom the - * Software is furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL - * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER - * DEALINGS IN THE SOFTWARE. - * - */ - -use std::time::Duration; - -use serde::Deserialize; - -/// Service Config uses the [protobuf Duration format](https://protobuf.dev/reference/protobuf/google.protobuf/#duration) -/// when parsing. The format is a string with a number followed -/// by an optional fraction and the letter 's' for seconds. For example, "1s", -/// "1.5s", "0.000000001s" are valid durations. -pub(crate) fn parse_duration(s: &str) -> Result { - if !s.ends_with('s') { - return Err("duration string must end with 's'".to_string()); - } - let s = &s[..s.len() - 1]; // strip 's'. - let mut parts = s.splitn(2, '.'); - let secs_str = parts - .next() - .ok_or_else(|| "empty duration string".to_string())?; - let secs: u64 = secs_str - .parse() - .map_err(|e| format!("failed to parse seconds: {}", e))?; - - let nanos = if let Some(fraction_str) = parts.next() { - if fraction_str.is_empty() { - return Err("empty fraction part".to_string()); - } - if fraction_str.len() > 9 { - return Err("fraction part has more than 9 digits".to_string()); - } - let fraction_val: u32 = fraction_str - .parse() - .map_err(|e| format!("failed to parse fraction: {}", e))?; - let pad = 9 - fraction_str.len(); - fraction_val * 10u32.pow(pad as u32) - } else { - 0 - }; - - Ok(Duration::new(secs, nanos)) -} - -pub(crate) fn deserialize_duration<'de, D>(deserializer: D) -> Result -where - D: serde::Deserializer<'de>, -{ - let s = String::deserialize(deserializer)?; - parse_duration(&s).map_err(serde::de::Error::custom) -} - -// Required to ensure that an optional Duraction can be parsed even when -// missing. -pub(crate) fn deserialize_duration_opt<'de, D>( - deserializer: D, -) -> Result, D::Error> -where - D: serde::Deserializer<'de>, -{ - let opt = Option::::deserialize(deserializer)?; - opt.map(|s| parse_duration(&s).map_err(serde::de::Error::custom)) - .transpose() -} - -// This is included to allow deserializing both u32 and string (or -// "stringified") representations of u32, as well as null values. This is used, -// for example, for the `max_request_message_bytes` field in the service config. -pub(crate) fn deserialize_uint32_opt<'de, D>(deserializer: D) -> Result, D::Error> -where - D: serde::Deserializer<'de>, -{ - struct OptVisitor; - impl<'de> serde::de::Visitor<'de> for OptVisitor { - type Value = Option; - - fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str("a u32 or a string representing a u32 or null") - } - - fn visit_none(self) -> Result - where - E: serde::de::Error, - { - Ok(None) - } - - fn visit_u64(self, v: u64) -> Result - where - E: serde::de::Error, - { - v.try_into().map(Some).map_err(serde::de::Error::custom) - } - - fn visit_str(self, v: &str) -> Result - where - E: serde::de::Error, - { - v.parse().map(Some).map_err(serde::de::Error::custom) - } - - fn visit_some(self, deserializer: D) -> Result - where - D: serde::Deserializer<'de>, - { - deserializer.deserialize_any(self) - } - } - deserializer.deserialize_option(OptVisitor) -} - -#[cfg(test)] -mod test { - use serde_json::json; - - use super::*; - - #[test] - fn test_parse_duration() { - assert_eq!(parse_duration("1s").unwrap(), Duration::from_secs(1)); - assert_eq!(parse_duration("0s").unwrap(), Duration::from_secs(0)); - assert_eq!( - parse_duration("1.5s").unwrap(), - Duration::new(1, 500_000_000) - ); - assert_eq!(parse_duration("0.000000001s").unwrap(), Duration::new(0, 1)); - assert_eq!(parse_duration("1.0000001s").unwrap(), Duration::new(1, 100)); - assert_eq!( - parse_duration("1.0001s").unwrap(), - Duration::new(1, 100_000) - ); - - assert!(parse_duration("").is_err()); - assert!(parse_duration("1").is_err()); - assert!(parse_duration("1.s").is_err()); - assert!(parse_duration("1.0000000001s").is_err()); - assert!(parse_duration("as").is_err()); - assert!(parse_duration(".5s").is_err()); - } - - #[test] - fn test_deserialize_duration() { - #[derive(Deserialize, Debug, PartialEq)] - struct TestStruct { - #[serde(deserialize_with = "deserialize_duration")] - d: Duration, - } - - let val: TestStruct = serde_json::from_value(json!({ "d": "1.5s" })).unwrap(); - assert_eq!(val.d, Duration::new(1, 500_000_000)); - - let res: Result = serde_json::from_value(json!({ "d": "1" })); - assert!(res.is_err()); - } - - #[test] - fn test_deserialize_duration_opt() { - #[derive(Deserialize, Debug, PartialEq)] - struct TestStruct { - #[serde(default, deserialize_with = "deserialize_duration_opt")] - d: Option, - } - - let val: TestStruct = serde_json::from_value(json!({ "d": "1.5s" })).unwrap(); - assert_eq!(val.d, Some(Duration::new(1, 500_000_000))); - - let val: TestStruct = serde_json::from_value(json!({ "d": null })).unwrap(); - assert_eq!(val.d, None); - - let val: TestStruct = serde_json::from_value(json!({})).unwrap(); - assert_eq!(val.d, None); - - let res: Result = serde_json::from_value(json!({ "d": "invalid" })); - assert!(res.is_err()); - } - - #[test] - fn test_deserialize_uint32_opt() { - #[derive(Deserialize, Debug, PartialEq)] - struct TestStruct { - #[serde(default, deserialize_with = "deserialize_uint32_opt")] - val: Option, - } - - let val: TestStruct = serde_json::from_value(json!({ "val": 123 })).unwrap(); - assert_eq!(val.val, Some(123)); - - let val: TestStruct = serde_json::from_value(json!({ "val": "456" })).unwrap(); - assert_eq!(val.val, Some(456)); - - let val: TestStruct = serde_json::from_value(json!({ "val": null })).unwrap(); - assert_eq!(val.val, None); - - let val: TestStruct = serde_json::from_value(json!({})).unwrap(); - assert_eq!(val.val, None); - - let res: Result = serde_json::from_value(json!({ "val": "invalid" })); - assert!(res.is_err()); - - let res: Result = serde_json::from_value(json!({ "val": -1 })); - assert!(res.is_err()); - } -} diff --git a/grpc/src/client/service_config/mod.rs b/grpc/src/client/service_config/mod.rs index aa139698a..efdb404d4 100644 --- a/grpc/src/client/service_config/mod.rs +++ b/grpc/src/client/service_config/mod.rs @@ -2,30 +2,33 @@ * * Copyright 2026 gRPC authors. * - * Permission is hereby granted, free of charge, to any person obtaining a - * copy of this software and associated documentation files (the "Software"), - * to deal in the Software without restriction, including without limitation - * the rights to use, copy, modify, merge, publish, distribute, sublicense, - * and/or sell copies of the Software, and to permit persons to whom the - * Software is furnished to do so, subject to the following conditions: + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL - * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER - * DEALINGS IN THE SOFTWARE. + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. * */ -use std::time::Duration; -pub(crate) mod json; +pub(crate) mod duration; pub(crate) mod serde; +use duration::GrpcDuration; + +pub type ParseResult = Result; + /// An in-memory representation of a service config, provided to gRPC as a JSON /// object. #[derive(Debug, Clone, PartialEq)] @@ -48,7 +51,7 @@ pub struct MethodConfig { /// List of methods this config applies to. pub(crate) name: Vec, pub(crate) wait_for_ready: Option, - pub(crate) timeout: Option, + pub(crate) timeout: Option, pub(crate) retry_policy: Option, pub(crate) hedging_policy: Option, pub max_request_message_bytes: Option, @@ -75,8 +78,8 @@ pub struct RetryThrottlingPolicy { #[derive(Debug, Clone, PartialEq)] pub struct RetryPolicy { pub(crate) max_attempts: u32, - pub(crate) initial_backoff: Duration, - pub(crate) max_backoff: Duration, + pub(crate) initial_backoff: GrpcDuration, + pub(crate) max_backoff: GrpcDuration, pub(crate) backoff_multiplier: f32, pub(crate) retryable_status_codes: Vec, } @@ -84,7 +87,7 @@ pub struct RetryPolicy { #[derive(Debug, Clone, PartialEq)] pub struct HedgingPolicy { pub(crate) max_attempts: u32, - pub(crate) hedging_delay: Duration, + pub(crate) hedging_delay: GrpcDuration, pub(crate) non_fatal_status_codes: Option>, } @@ -123,56 +126,20 @@ impl MethodConfig { } } -#[derive(Debug, Clone, PartialEq)] -pub struct ParseResult(pub Result); - -impl ParseResult { - pub fn is_ok(&self) -> bool { - self.0.is_ok() - } - - pub fn is_err(&self) -> bool { - self.0.is_err() - } - - pub fn unwrap(self) -> ServiceConfig { - self.0.unwrap() - } - - pub fn unwrap_err(self) -> String { - self.0.unwrap_err() - } - - pub fn into_inner(self) -> Result { - self.0 - } -} - -impl std::ops::Deref for ParseResult { - type Target = Result; - - fn deref(&self) -> &Self::Target { - &self.0 - } -} - impl ServiceConfig { /// Parses a service configuration from a JSON string. pub fn parse(config_json: &str) -> ParseResult { let config_serde: serde::ServiceConfigSerDe = match serde_json::from_str(config_json) { Ok(c) => c, Err(e) => { - return ParseResult(Err(format!( - "failed to deserialize service config JSON: {}", - e - ))); + return Err(format!("failed to deserialize service config JSON: {}", e)); } }; let config: Self = config_serde.into(); if let Err(e) = config.validate() { - return ParseResult(Err(e)); + return Err(e); } - ParseResult(Ok(config)) + Ok(config) } fn validate(&self) -> Result<(), String> { @@ -247,6 +214,8 @@ impl ServiceConfig { #[cfg(test)] mod test { + use std::time::Duration; + use serde_json::json; use super::*; @@ -319,15 +288,21 @@ mod test { assert_eq!(mc.name[1].service, "grpc.examples.echo.Echo2"); assert_eq!(mc.name[1].method, None); - assert_eq!(mc.timeout, Some(Duration::new(1, 500_000_000))); + assert_eq!( + mc.timeout, + Some(GrpcDuration(Duration::new(1, 500_000_000))) + ); assert_eq!(mc.max_request_message_bytes, Some(1024)); assert_eq!(mc.max_response_message_bytes, Some(2048)); // Verify Retry Policy. let rp = mc.retry_policy.as_ref().unwrap(); assert_eq!(rp.max_attempts, 3); - assert_eq!(rp.initial_backoff, Duration::new(0, 100_000_000)); - assert_eq!(rp.max_backoff, Duration::from_secs(1)); + assert_eq!( + rp.initial_backoff, + GrpcDuration(Duration::new(0, 100_000_000)) + ); + assert_eq!(rp.max_backoff, GrpcDuration(Duration::from_secs(1))); assert_eq!(rp.backoff_multiplier, 2.0); assert_eq!(rp.retryable_status_codes, vec!["UNAVAILABLE", "INTERNAL"]); @@ -335,7 +310,7 @@ mod test { assert_eq!(mc2.wait_for_ready, Some(true)); let hp = mc2.hedging_policy.as_ref().unwrap(); assert_eq!(hp.max_attempts, 3); - assert_eq!(hp.hedging_delay, Duration::from_millis(500)); + assert_eq!(hp.hedging_delay, GrpcDuration(Duration::from_millis(500))); assert_eq!( hp.non_fatal_status_codes, Some(vec!["UNAVAILABLE".to_string()]) diff --git a/grpc/src/client/service_config/serde.rs b/grpc/src/client/service_config/serde.rs index f1afe3400..6ac1399bd 100644 --- a/grpc/src/client/service_config/serde.rs +++ b/grpc/src/client/service_config/serde.rs @@ -1,4 +1,26 @@ -use std::time::Duration; +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ use serde::Deserialize; use serde::Serialize; @@ -12,7 +34,7 @@ use super::MethodName; use super::RetryPolicy; use super::RetryThrottlingPolicy; use super::ServiceConfig; -use super::json; +use super::duration::GrpcDuration; #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] @@ -30,13 +52,10 @@ pub(crate) struct ServiceConfigSerDe { pub(crate) struct MethodConfigSerDe { pub(crate) name: Vec, pub(crate) wait_for_ready: Option, - #[serde(default, deserialize_with = "json::deserialize_duration_opt")] - pub(crate) timeout: Option, + pub(crate) timeout: Option, pub(crate) retry_policy: Option, pub(crate) hedging_policy: Option, - #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] pub(crate) max_request_message_bytes: Option, - #[serde(default, deserialize_with = "json::deserialize_uint32_opt")] pub(crate) max_response_message_bytes: Option, } @@ -57,10 +76,8 @@ pub(crate) struct RetryThrottlingPolicySerDe { #[serde(rename_all = "camelCase")] pub(crate) struct RetryPolicySerDe { pub(crate) max_attempts: u32, - #[serde(deserialize_with = "json::deserialize_duration")] - pub(crate) initial_backoff: Duration, - #[serde(deserialize_with = "json::deserialize_duration")] - pub(crate) max_backoff: Duration, + pub(crate) initial_backoff: GrpcDuration, + pub(crate) max_backoff: GrpcDuration, pub(crate) backoff_multiplier: f32, pub(crate) retryable_status_codes: Vec, } @@ -69,8 +86,7 @@ pub(crate) struct RetryPolicySerDe { #[serde(rename_all = "camelCase")] pub(crate) struct HedgingPolicySerDe { pub(crate) max_attempts: u32, - #[serde(deserialize_with = "json::deserialize_duration")] - pub(crate) hedging_delay: Duration, + pub(crate) hedging_delay: GrpcDuration, #[serde(default)] pub(crate) non_fatal_status_codes: Option>, } @@ -174,7 +190,7 @@ impl From for MethodConfig { Self { name: dto.name.into_iter().map(Into::into).collect(), wait_for_ready: dto.wait_for_ready, - timeout: dto.timeout, + timeout: dto.timeout.map(Into::into), retry_policy: dto.retry_policy.map(Into::into), hedging_policy: dto.hedging_policy.map(Into::into), max_request_message_bytes: dto.max_request_message_bytes, diff --git a/grpc/src/inmemory/mod.rs b/grpc/src/inmemory/mod.rs index 38a86052a..41f40b624 100644 --- a/grpc/src/inmemory/mod.rs +++ b/grpc/src/inmemory/mod.rs @@ -374,7 +374,8 @@ impl Transport for InMemoryTransport { } } -/// An implementation of [`ClientConnectionSecurityContext`] for in-memory connections. +/// An implementation of [`ClientConnectionSecurityContext`] for in-memory +/// connections. #[derive(Debug, Clone)] struct InMemoryChannelecurityContext; @@ -426,7 +427,6 @@ impl Resolver for InMemoryResolver { let service_config = channel_controller .parse_service_config(r#"{"loadBalancingConfig": [{"round_robin": {}}]}"#) - .into_inner() .map(Some); let _ = channel_controller.update(ResolverUpdate { From 0a4befa5639cf3975946012ea478a4ad49f31b75 Mon Sep 17 00:00:00 2001 From: Nathaniel Ford Date: Wed, 22 Jul 2026 20:21:56 +0000 Subject: [PATCH 6/6] Restore deserialize_uint32_opt in serde.rs TAG=agy CONV=5f23221a-0150-4021-ad9c-111618d24771 --- grpc/src/client/service_config/serde.rs | 45 +++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/grpc/src/client/service_config/serde.rs b/grpc/src/client/service_config/serde.rs index 6ac1399bd..1ebffc4b9 100644 --- a/grpc/src/client/service_config/serde.rs +++ b/grpc/src/client/service_config/serde.rs @@ -55,7 +55,9 @@ pub(crate) struct MethodConfigSerDe { pub(crate) timeout: Option, pub(crate) retry_policy: Option, pub(crate) hedging_policy: Option, + #[serde(default, deserialize_with = "deserialize_uint32_opt")] pub(crate) max_request_message_bytes: Option, + #[serde(default, deserialize_with = "deserialize_uint32_opt")] pub(crate) max_response_message_bytes: Option, } @@ -263,3 +265,46 @@ impl From for LoadBalancingConfig { } } } + +pub(crate) fn deserialize_uint32_opt<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + struct OptVisitor; + impl<'de> serde::de::Visitor<'de> for OptVisitor { + type Value = Option; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("a u32 or a string representing a u32 or null") + } + + fn visit_none(self) -> Result + where + E: serde::de::Error, + { + Ok(None) + } + + fn visit_u64(self, v: u64) -> Result + where + E: serde::de::Error, + { + v.try_into().map(Some).map_err(serde::de::Error::custom) + } + + fn visit_str(self, v: &str) -> Result + where + E: serde::de::Error, + { + v.parse().map(Some).map_err(serde::de::Error::custom) + } + + fn visit_some(self, deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_any(self) + } + } + deserializer.deserialize_option(OptVisitor) +}