diff --git a/grpc/src/client/channel.rs b/grpc/src/client/channel.rs index 87928b00d..65f81946e 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. * */ @@ -65,8 +65,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::LbPolicyType; -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; @@ -250,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 { @@ -268,8 +268,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(); @@ -283,8 +284,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 { @@ -448,9 +451,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 { @@ -466,7 +469,7 @@ impl name_resolution::ChannelController for ResolverChannelController { .map_err(|err| err.to_string()) } - fn parse_service_config(&self, config: &str) -> Result { + fn parse_service_config(&self, config: &str) -> 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..d0f452317 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. * */ @@ -42,6 +42,7 @@ 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)] 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..e2b116162 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 { + fn parse_service_config(&self, _: &str) -> ParseResult { Err("Unimplemented".to_string()) } } 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/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/mod.rs b/grpc/src/client/service_config/mod.rs new file mode 100644 index 000000000..efdb404d4 --- /dev/null +++ b/grpc/src/client/service_config/mod.rs @@ -0,0 +1,412 @@ +/* + * + * 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. + * + */ + +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)] +pub struct ServiceConfig { + pub(crate) load_balancing_policy: Option, + /// Ordered list of LB policies. The first supported one is used. + pub(crate) load_balancing_config: Option>, + /// Per-method configuration overrides. + pub(crate) method_config: Option>, + /// Global retry throttling parameters. + 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, PartialEq)] +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) retry_policy: Option, + pub(crate) hedging_policy: Option, + pub max_request_message_bytes: Option, + pub(crate) max_response_message_bytes: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MethodName { + /// The service name (fully qualified). + /// 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. + /// e.g., for "grpc.examples.echo.Echo/Echo" this would be "Echo". + pub(crate) method: Option, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RetryThrottlingPolicy { + pub(crate) max_tokens: u32, + pub(crate) token_ratio: f32, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RetryPolicy { + pub(crate) max_attempts: u32, + pub(crate) initial_backoff: GrpcDuration, + pub(crate) max_backoff: GrpcDuration, + 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: GrpcDuration, + 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(crate) name: String, + pub(crate) config: serde_json::Value, +} + +#[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 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 Err(format!("failed to deserialize service config JSON: {}", e)); + } + }; + let config: Self = config_serde.into(); + if let Err(e) = config.validate() { + return Err(e); + } + 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 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!( + "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 hp) = mc.hedging_policy + && hp.max_attempts <= 1 + { + return Err(format!( + "method_config[{}].hedging_policy.max_attempts must be > 1", + 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 std::time::Duration; + + use serde_json::json; + + use super::*; + + #[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" + }, + { + "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 + } + }); + + 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(), 2); + 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(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, + 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"]); + + 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, GrpcDuration(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] + 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()); + } + + #[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..1ebffc4b9 --- /dev/null +++ b/grpc/src/client/service_config/serde.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; +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::duration::GrpcDuration; + +#[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, + 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, +} + +#[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, + pub(crate) initial_backoff: GrpcDuration, + pub(crate) max_backoff: GrpcDuration, + 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, + pub(crate) hedging_delay: GrpcDuration, + #[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.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, + 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, + } + } +} + +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) +} diff --git a/grpc/src/inmemory/mod.rs b/grpc/src/inmemory/mod.rs index 4bba7357f..41f40b624 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; @@ -375,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; @@ -425,13 +425,13 @@ impl Resolver for InMemoryResolver { }) .collect(); + let service_config = channel_controller + .parse_service_config(r#"{"loadBalancingConfig": [{"round_robin": {}}]}"#) + .map(Some); + let _ = channel_controller.update(ResolverUpdate { endpoints: Ok(endpoints), - service_config: Ok(Some(ServiceConfig { - load_balancing_policy: Some( - crate::client::service_config::LbPolicyType::RoundRobin, - ), - })), + service_config, ..Default::default() }); }