From 65b6b20c28cef710b4ff10a7671ff22388b203b6 Mon Sep 17 00:00:00 2001 From: Yu Liu <60283975+LYZJU2019@users.noreply.github.com> Date: Fri, 31 Jul 2026 12:12:50 -0700 Subject: [PATCH 1/4] feat(tonic-xds): drive gRPC retry config from RDS RouteAction.retry_policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the gRPC channel's retry configuration to the control plane so retry behavior tracks RDS (RouteConfiguration) updates without rebuilding the channel. The retry settings come from the standard Envoy `RouteAction.retry_policy` (gRFC A44), so OSS parses them natively — no caller-supplied extractor is needed. Retry is per route: each request retries according to the exact route it matched. - Parse and validate the retry policy when the RDS resource is validated — once, not per request. `validate_route` maps the Envoy `retry_on` conditions to gRPC status codes, applies defaults for unset fields, and stores the result as an immutable, `Arc`-shared `GrpcRetrySharedConfig` on the matched route (`RouteConfig.retry_config`). A route uses its own `RouteAction.retry_policy` when set, otherwise it inherits the enclosing `VirtualHost.retry_policy` (gRFC A44: a route-level policy completely overrides the virtual host's — values are not merged). `None` when neither specifies retry; routes inheriting the vhost policy share one `Arc`. `RouteRetryConfig` remains a transport-neutral carrier for the raw Envoy values during parsing. - Select the config per route via the routing decision. The routing layer stamps the matched route's shared retry config into the request's `RouteDecision`, taken from the same config snapshot it routed with. Because routing and retry read one snapshot, they always act on the same RDS version (no cross-layer skew), and retry runs inside routing so the decision is fixed across a request's retry attempts. - Separate the shared, immutable retry config (attempt cap, backoff, retryable code set) from the per-request retry state (backoff cursor, attempt count). `RetrySharedConfig` holds the config behind an `Arc`; `RetryPolicy` holds a pointer to it plus the per-request state. Instantiating a policy for a request (`RetryPolicy::from_shared`) is an `Arc` pointer clone plus a zero-field state init — the request hot path does no parsing or allocation. Requests with no `RouteDecision` (non-xDS callers) or whose route carries no retry policy use the layer's fallback config. - Map Envoy `retry_on` conditions to gRPC status codes (gRFC A44) with `grpc_retry_on_codes`; non-gRPC tokens are ignored (connection-level retries are handled separately). Envoy `numRetries` maps directly to `RetryConfig.num_retries` (retries, not attempts). The retry engine (`RetryPolicy`, `RetrySharedConfig`, and the `RetryClassifier` seam) stays transport-agnostic; only the retry layer is gRPC-specific, because it reads the concrete `RouteDecision` extension. Deriving and validating the gRPC retry config once at RDS-validation time — rather than in the transport- neutral resource type — is what keeps the request path allocation-free. --- tonic-xds/src/client/channel.rs | 20 +- tonic-xds/src/client/circuit_breaking.rs | 5 +- .../client/loadbalance/pickers/ring_hash.rs | 1 + tonic-xds/src/client/retry.rs | 373 +++++++++++++----- tonic-xds/src/client/route.rs | 13 + tonic-xds/src/xds/cache.rs | 1 + tonic-xds/src/xds/resource/route_config.rs | 199 +++++++++- tonic-xds/src/xds/resource_manager.rs | 2 + tonic-xds/src/xds/routing.rs | 48 ++- 9 files changed, 554 insertions(+), 108 deletions(-) diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 7b4b19c73..ea1133895 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -367,6 +367,18 @@ impl XdsChannelBuilder { resource_manager: XdsResourceManager, ) -> XdsChannelGrpc { let router: Arc = Arc::new(XdsRouter::new(&cache)); + + // Retry config is control-plane-driven from RDS, per route. It is parsed + // and validated once, when the `RouteConfiguration` is validated. The + // routing layer (outer) stamps the matched route's shared retry config + // into the request's `RouteDecision`; the retry layer reads the shared + // config `Arc` and instantiates a per-request policy from it, so the + // request hot path does no parsing or allocation. The default below is + // the fallback used when a request carries no route retry config (non-xDS + // callers, or a route with no retry policy). See + // [`RetryLayer`](crate::client::retry::RetryLayer). + let retry_layer = RetryLayer::new(GrpcRetryPolicy::default()); + #[cfg(feature = "_tls-any")] let discovery: Arc< dyn ClusterDiscovery>, @@ -378,7 +390,6 @@ impl XdsChannelBuilder { let discovery: Arc< dyn ClusterDiscovery>, > = Arc::new(XdsClusterDiscovery::new(cache, GrpcMakeConnector::new())); - let retry_policy = GrpcRetryPolicy::default(); let resources = Arc::new(XdsChannelResources { _resource_manager: resource_manager, @@ -386,7 +397,6 @@ impl XdsChannelBuilder { }); let routing_layer = XdsRoutingLayer::new(router, self.pre_route.clone(), self.authority()); - let retry_layer = RetryLayer::new(retry_policy); let cluster_registry = Arc::new(ClusterClientRegistryGrpc::new()); let lb_service = XdsLbService::new(cluster_registry, discovery); let inner = ServiceBuilder::new() @@ -527,6 +537,7 @@ mod tests { Ok(RouteDecision { cluster: "test-cluster".to_string(), request_hash: None, + retry_config: None, }) } } @@ -680,9 +691,7 @@ mod tests { let retry_policy = GrpcRetryPolicy::new( RetryConfig::new().num_retries(1), - GrpcRetryClassifier { - retry_on: vec![tonic::Code::Unavailable], - }, + GrpcRetryClassifier::new(vec![tonic::Code::Unavailable]), ); let xds_channel = XdsChannelBuilder::new(test_config()).build_grpc_channel_from_parts( @@ -734,6 +743,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster(cluster_name.to_string()), + retry_config: None, }], }], metadata: Default::default(), diff --git a/tonic-xds/src/client/circuit_breaking.rs b/tonic-xds/src/client/circuit_breaking.rs index 1805acabe..3b5a60029 100644 --- a/tonic-xds/src/client/circuit_breaking.rs +++ b/tonic-xds/src/client/circuit_breaking.rs @@ -731,6 +731,7 @@ mod tests { request.extensions_mut().insert(RouteDecision { cluster: CLUSTER.to_string(), request_hash: None, + retry_config: None, }); request } @@ -984,9 +985,7 @@ mod tests { RetryBackoffConfig::new(Duration::from_millis(1)) .max_interval(Duration::from_millis(1)), ), - GrpcRetryClassifier { - retry_on: vec![Code::Unavailable], - }, + GrpcRetryClassifier::new(vec![Code::Unavailable]), ); let calls = Arc::new(AtomicU32::new(0)); let call_counter = calls.clone(); diff --git a/tonic-xds/src/client/loadbalance/pickers/ring_hash.rs b/tonic-xds/src/client/loadbalance/pickers/ring_hash.rs index 68dc881d9..77f65ce13 100644 --- a/tonic-xds/src/client/loadbalance/pickers/ring_hash.rs +++ b/tonic-xds/src/client/loadbalance/pickers/ring_hash.rs @@ -223,6 +223,7 @@ mod tests { r.extensions_mut().insert(RouteDecision { cluster: "c".to_string(), request_hash: hash, + retry_config: None, }); r } diff --git a/tonic-xds/src/client/retry.rs b/tonic-xds/src/client/retry.rs index 2d0060deb..8e718ee07 100644 --- a/tonic-xds/src/client/retry.rs +++ b/tonic-xds/src/client/retry.rs @@ -36,7 +36,6 @@ use std::sync::Arc; use std::task::{Context, Poll}; use std::time::Duration; -use arc_swap::ArcSwap; use backoff::ExponentialBackoffBuilder; use backoff::backoff::Backoff; use http::{Request, Response}; @@ -47,6 +46,8 @@ use tower::retry::Retry; use tower::{Layer, Service}; use crate::client::circuit_breaking::is_local_circuit_breaker_drop; +use crate::client::route::RouteDecision; +use crate::xds::resource::route_config::RouteRetryConfig; /// Check if an error's source chain contains a retryable connection-level error. /// @@ -205,10 +206,30 @@ impl Default for RetryConfig { /// Default gRPC [`RetryClassifier`]: retries on a retryable connection error or a /// retryable gRPC status code, and stamps the `grpc-previous-rpc-attempts` header /// on each retry per the gRPC spec. -#[derive(Debug, Clone, Default)] +/// +/// The retryable gRPC status codes are parsed once per `RouteConfiguration` and +/// held behind an `Arc<[Code]>`, so cloning a classifier (per request, or on each +/// tower retry) is a pointer bump rather than a re-parse or re-allocation. A new +/// set is built only when RDS changes; see [`RetryLayer`]. +#[derive(Debug, Clone)] pub(crate) struct GrpcRetryClassifier { - /// gRPC status codes that should be retried. - pub(crate) retry_on: Vec, + /// gRPC status codes that should be retried. Shared (parsed once per config). + retry_on: Arc<[tonic::Code]>, +} + +impl GrpcRetryClassifier { + /// Create a classifier that retries the given gRPC status codes. + pub(crate) fn new(retry_on: Vec) -> Self { + Self { + retry_on: retry_on.into(), + } + } +} + +impl Default for GrpcRetryClassifier { + fn default() -> Self { + Self::new(Vec::new()) + } } impl RetryClassifier for GrpcRetryClassifier { @@ -245,24 +266,56 @@ fn make_backoff(config: &RetryBackoffConfig) -> backoff::ExponentialBackoff { .build() } -/// Retry *decision* state — attempt cap, backoff, and body cloning — shared -/// across transports. Transport-specific decisions live in the classifier `C` +/// Immutable, shared retry configuration — the transport-agnostic knobs +/// (attempt cap, backoff) plus the transport-specific classifier `C`. Built and +/// validated **once** when a `RouteConfiguration` is validated (see +/// [`GrpcRetrySharedConfig::from_route_retry`]) and shared across every request +/// that matches the route via an [`Arc`], so per-request setup never re-parses +/// or re-allocates. +/// +/// Kept separate from the per-request retry *state* ([`RetryPolicy`]) so that +/// instantiating a policy for a request is just an `Arc` pointer clone plus a +/// zero-field state init — see [`RetryPolicy::from_shared`]. +#[derive(Debug)] +pub(crate) struct RetrySharedConfig { + /// Attempt cap and backoff schedule, shared by every [`RetryClassifier`]. + config: RetryConfig, + /// Decides retryability and per-retry request mutation for the transport. + classifier: C, +} + +impl RetrySharedConfig { + /// Create a shared retry config from an attempt/backoff [`RetryConfig`] and a + /// transport classifier. + pub(crate) fn new(config: RetryConfig, classifier: C) -> Self { + Self { config, classifier } + } +} + +impl Default for RetrySharedConfig { + fn default() -> Self { + Self::new(RetryConfig::default(), C::default()) + } +} + +/// Per-request retry *state* — a pointer to the shared, immutable +/// [`RetrySharedConfig`] plus the mutable state for one request (backoff cursor +/// and attempt counter). Transport-specific decisions live in the classifier `C` /// (see [`RetryClassifier`]). /// -/// Wraps a [`RetryConfig`] behind an [`ArcSwap`] so that configuration can be -/// atomically updated (e.g. from xDS) without blocking in-flight requests. +/// Instantiating one for a request ([`RetryPolicy::from_shared`]) is a hot-path +/// operation: it clones a single `Arc` (a pointer bump) and zero-inits the state +/// fields — no parsing, allocation, or config copying. /// /// Implements [`tower::retry::Policy`]. Tower's `Retry` service clones the policy /// for each request, so `backoff` and `attempts` track per-request retry state -/// while the shared config is read from `ArcSwap` on each retry decision. The -/// retry *state machine* stays entirely in tower's `Retry`/`ResponseFuture`; -/// this type only implements the `Policy` trait, so no state machine is -/// reimplemented here. +/// while the shared config stays behind the `Arc`. The retry *state machine* +/// stays entirely in tower's `Retry`/`ResponseFuture`; this type only implements +/// the `Policy` trait, so no state machine is reimplemented here. #[derive(Clone, Debug)] pub(crate) struct RetryPolicy { - config: Arc>, - /// Decides retryability and per-retry request mutation for the transport. - classifier: C, + /// Immutable config shared across all requests on this route (pointer clone). + shared: Arc>, /// Backoff state for the current request, created from config on first retry. backoff: Option, /// Number of retry attempts made so far for the current request. @@ -270,28 +323,32 @@ pub(crate) struct RetryPolicy { } impl RetryPolicy { - /// Create a new retry policy with the given configuration and classifier. + /// Create a policy from a shared config and a classifier, allocating a fresh + /// [`Arc`] for the shared config. Prefer [`from_shared`](Self::from_shared) + /// on the hot path, where the shared config already lives behind an `Arc`. pub(crate) fn new(config: RetryConfig, classifier: C) -> Self { + Self::from_shared(Arc::new(RetrySharedConfig::new(config, classifier))) + } + + /// Instantiate per-request retry state from an already-shared config. + /// + /// Hot path: a single `Arc` pointer clone (the caller's) plus a zero-field + /// state init. No parsing, allocation, or config copy. + pub(crate) fn from_shared(shared: Arc>) -> Self { Self { - config: Arc::new(ArcSwap::from(Arc::new(config))), - classifier, + shared, backoff: None, attempts: 0, } } - /// Atomically swap the configuration with a new one. - pub(crate) fn update_config(&self, config: RetryConfig) { - self.config.store(Arc::new(config)); - } - - /// Load the current configuration. - pub(crate) fn load_config(&self) -> Arc { - self.config.load_full() - } - /// Get or create the backoff, and advance it to the next delay. - fn backoff_next(&mut self, backoff_config: &RetryBackoffConfig) -> Duration { + /// + /// Only called on the cold path (when a request is actually being retried), + /// never for successful requests. Borrows `shared` and `backoff` disjointly, + /// so it reads the shared backoff config by reference without cloning. + fn backoff_next(&mut self) -> Duration { + let backoff_config = &self.shared.config.retry_backoff; let backoff = self .backoff .get_or_insert_with(|| make_backoff(backoff_config)); @@ -303,10 +360,61 @@ impl RetryPolicy { impl Default for RetryPolicy { fn default() -> Self { - Self::new(RetryConfig::default(), C::default()) + Self::from_shared(Arc::new(RetrySharedConfig::default())) } } +impl RetrySharedConfig { + /// Build a shared gRPC retry config from a route's [`RouteRetryConfig`] (RDS + /// `RouteAction.retry_policy`). Unset Envoy fields fall back to + /// [`RetryConfig`] defaults. + /// + /// This is where `retry_on` is parsed into [`tonic::Code`]s and the + /// retryable-code set is allocated, so it runs **once, when the + /// `RouteConfiguration` is validated** — not per request. The result is + /// wrapped in an [`Arc`] and carried on the matched route, so instantiating a + /// per-request policy is just a pointer clone (see + /// [`RetryPolicy::from_shared`]). + pub(crate) fn from_route_retry(retry: &RouteRetryConfig) -> Self { + let mut config = RetryConfig::new(); + if let Some(num_retries) = retry.num_retries { + config = config.num_retries(num_retries); + } + if let Some(base_interval) = retry.base_interval { + let mut backoff = RetryBackoffConfig::new(base_interval); + if let Some(max_interval) = retry.max_interval { + backoff = backoff.max_interval(max_interval); + } + config = config.retry_backoff(backoff); + } + Self::new( + config, + GrpcRetryClassifier::new(grpc_retry_on_codes(&retry.retry_on)), + ) + } +} + +/// Map Envoy `retry_on` conditions (comma-separated) to gRPC [`tonic::Code`]s. +/// +/// Only the gRPC-status conditions from gRFC A44 are recognized; non-gRPC tokens +/// (e.g. `5xx`, `gateway-error`, `reset`, `connect-failure`) are ignored because +/// connection-level retries are handled separately by +/// [`is_retryable_connection_error`]. +pub(crate) fn grpc_retry_on_codes(retry_on: &str) -> Vec { + use tonic::Code; + retry_on + .split(',') + .filter_map(|token| match token.trim() { + "cancelled" => Some(Code::Cancelled), + "deadline-exceeded" => Some(Code::DeadlineExceeded), + "internal" => Some(Code::Internal), + "resource-exhausted" => Some(Code::ResourceExhausted), + "unavailable" => Some(Code::Unavailable), + _ => None, + }) + .collect() +} + impl Policy, Response, tower::BoxError> for RetryPolicy where C: RetryClassifier, @@ -319,22 +427,20 @@ where req: &mut Request, result: &mut Result, tower::BoxError>, ) -> Option { - let config = self.load_config(); - - if self.attempts >= config.num_retries { + if self.attempts >= self.shared.config.num_retries { return None; } - if !self.classifier.is_retryable(result) { + if !self.shared.classifier.is_retryable(result) { return None; } - let delay = self.backoff_next(&config.retry_backoff); + let delay = self.backoff_next(); self.attempts += 1; // Let the classifier stamp any per-retry request state (e.g. gRPC's // grpc-previous-rpc-attempts header). - self.classifier.prepare_retry(req, self.attempts); + self.shared.classifier.prepare_retry(req, self.attempts); Some(tokio::time::sleep(delay)) } @@ -347,50 +453,92 @@ where /// Non-breaking alias: existing gRPC callers keep the same name and behavior. pub(crate) type GrpcRetryPolicy = RetryPolicy; -/// Tower [`Layer`] that wraps a service with retry support. +/// Shared, immutable gRPC retry config (attempt cap, backoff, and the retryable +/// [`tonic::Code`] set). Built and validated once per `RouteConfiguration` (see +/// [`RetrySharedConfig::from_route_retry`]) and carried behind an [`Arc`] on the +/// matched route, so instantiating a per-request [`GrpcRetryPolicy`] is a pointer +/// clone (see [`RetryPolicy::from_shared`]). +pub(crate) type GrpcRetrySharedConfig = RetrySharedConfig; + +/// Tower [`Layer`] that wraps a gRPC service with retry support. /// /// Converts the request body into a [`SharedBody`] (cloneable) and constructs /// a fresh [`tower::retry::Retry`] service per request so that each request /// gets its own retry state. /// -/// This layer is generic over the retry policy — it is not tied to gRPC. -/// The gRPC-specific behavior lives in the [`Policy`] implementation -/// (e.g. [`GrpcRetryPolicy`]). +/// # Per-route, control-plane-driven policy +/// +/// The active retry config is selected **per request** from the route the +/// request matched. The routing layer (which runs immediately outside this one) +/// stamps the matched route's shared retry config into the request's +/// [`RouteDecision`], taken from the same config snapshot it routed with. Because +/// both layers read one snapshot, routing and retry always act on the same RDS +/// version — there is no cross-layer version skew — and each request retries +/// according to the exact route it took (gRFC A44). +/// +/// Requests with no [`RouteDecision`] (non-xDS callers) or whose matched route +/// carries no retry policy use `fallback`. +/// +/// # Hot path +/// +/// The gRPC retry config (`retry_on` codes, attempt cap, backoff) is parsed and +/// validated **once**, when the `RouteConfiguration` is validated, and shared +/// behind an [`Arc`] on the matched route (see [`GrpcRetrySharedConfig`]). This +/// layer never parses or derives on the request path: it reads the shared config +/// `Arc` from the [`RouteDecision`], clones the pointer, and instantiates a +/// [`GrpcRetryPolicy`] with fresh per-request state (see +/// [`RetryPolicy::from_shared`]). No parsing, allocation, or locking. +/// +/// This layer is gRPC-specific because it reads the concrete [`RouteDecision`] +/// extension; the retry *engine* ([`RetryPolicy`], [`RetrySharedConfig`], and the +/// [`RetryClassifier`] seam) stays transport-agnostic. #[derive(Clone)] -pub(crate) struct RetryLayer

{ - policy: P, +pub(crate) struct RetryLayer { + /// Shared config used when a request carries no per-route retry config + /// (non-xDS callers, or a matched route with no `RouteAction.retry_policy`). + fallback: Arc, } -impl

RetryLayer

{ - /// Create a new retry layer with the given policy. - pub(crate) fn new(policy: P) -> Self { - Self { policy } +impl RetryLayer { + /// Create a retry layer whose `fallback` policy is used when a request + /// carries no per-route retry config. Only the policy's shared config is + /// kept; its per-request state is discarded. Route-specific configs are read + /// from the request's [`RouteDecision`] (see [`RetryLayer`]). + pub(crate) fn new(fallback: GrpcRetryPolicy) -> Self { + Self { + fallback: fallback.shared, + } } } -impl Layer for RetryLayer

{ - type Service = RetryService; +impl Layer for RetryLayer { + type Service = RetryService; fn layer(&self, service: S) -> Self::Service { RetryService { inner: service, - policy: self.policy.clone(), + fallback: Arc::clone(&self.fallback), } } } /// Service that converts request bodies to [`SharedBody`] and retries via -/// [`tower::retry::Retry`] with the given policy. +/// [`tower::retry::Retry`]. The retry config is selected per request from the +/// route the request matched (via [`RouteDecision`]) and is already parsed and +/// shared behind an [`Arc`], so the hot path performs no parsing or allocation +/// (see [`RetryLayer`]). #[derive(Clone)] -pub(crate) struct RetryService { +pub(crate) struct RetryService { inner: S, - policy: P, + /// Shared config used when a request carries no per-route retry config, + /// shared (via `Arc`) with the layer and every per-request clone. + fallback: Arc, } -impl Service> for RetryService +impl Service> for RetryService where - P: Policy>, Response, S::Error> + Clone + Send + 'static, - P::Future: Send, + GrpcRetryPolicy: Policy>, Response, S::Error>, + >, Response, S::Error>>::Future: Send, S: Service>, Response = Response> + Clone + Send + 'static, S::Error: Debug + Send + 'static, S::Response: Send + 'static, @@ -411,7 +559,20 @@ where } fn call(&mut self, request: Request) -> Self::Future { - let mut retry_svc = Retry::new(self.policy.clone(), self.inner.clone()); + // Select the shared retry config for the route this request matched. The + // routing layer (just outside this one) stamped the matched route's shared + // config into the `RouteDecision` from the same config snapshot it routed + // with. The config is already parsed and validated (done once when the + // `RouteConfiguration` was validated), so this is a pointer clone plus a + // zero-field state init — no parsing or allocation. Fall back when the + // request carries no route retry config. + let shared = request + .extensions() + .get::() + .and_then(|decision| decision.retry_config.clone()) + .unwrap_or_else(|| Arc::clone(&self.fallback)); + let policy = RetryPolicy::from_shared(shared); + let mut retry_svc = Retry::new(policy, self.inner.clone()); let shared_request = request.map(|b| b.into_shared()); Box::pin(retry_svc.call(shared_request)) } @@ -527,9 +688,7 @@ mod tests { #[test] fn test_is_retryable_grpc_status_via_result() { - let classifier = GrpcRetryClassifier { - retry_on: vec![tonic::Code::Unavailable], - }; + let classifier = GrpcRetryClassifier::new(vec![tonic::Code::Unavailable]); let response = http::Response::builder() .header("grpc-status", "14") // UNAVAILABLE .body(()) @@ -540,9 +699,7 @@ mod tests { #[test] fn test_is_not_retryable_ok_response() { - let classifier = GrpcRetryClassifier { - retry_on: vec![tonic::Code::Unavailable], - }; + let classifier = GrpcRetryClassifier::new(vec![tonic::Code::Unavailable]); let response = http::Response::builder() .header("grpc-status", "0") // OK .body(()) @@ -553,9 +710,7 @@ mod tests { #[test] fn test_is_not_retryable_no_grpc_status_header() { - let classifier = GrpcRetryClassifier { - retry_on: vec![tonic::Code::Unavailable], - }; + let classifier = GrpcRetryClassifier::new(vec![tonic::Code::Unavailable]); let response = http::Response::builder().body(()).unwrap(); let result: Result, tower::BoxError> = Ok(response); assert!(!classifier.is_retryable(&result)); @@ -635,12 +790,11 @@ mod tests { #[test] fn test_grpc_classifier_retry_on() { - let classifier = GrpcRetryClassifier { - retry_on: vec![tonic::Code::Unavailable, tonic::Code::Cancelled], - }; + let classifier = + GrpcRetryClassifier::new(vec![tonic::Code::Unavailable, tonic::Code::Cancelled]); assert_eq!( - classifier.retry_on, - vec![tonic::Code::Unavailable, tonic::Code::Cancelled] + classifier.retry_on.as_ref(), + [tonic::Code::Unavailable, tonic::Code::Cancelled] ); } @@ -653,29 +807,74 @@ mod tests { assert_eq!(config.retry_backoff, backoff); } - // --- RetryPolicy (ArcSwap wrapper) tests --- + // --- Building a policy from a route's retry config --- #[test] - fn test_policy_load_config() { - let policy = GrpcRetryPolicy::new( - RetryConfig::new().num_retries(1), - GrpcRetryClassifier { - retry_on: vec![tonic::Code::Unavailable], - }, + fn test_from_route_retry_maps_fields() { + let retry = RouteRetryConfig { + retry_on: "unavailable".into(), + num_retries: Some(3), + base_interval: Some(Duration::from_millis(100)), + max_interval: Some(Duration::from_millis(1000)), + }; + let shared = GrpcRetrySharedConfig::from_route_retry(&retry); + assert_eq!(shared.config.num_retries, 3); + assert_eq!( + shared.config.retry_backoff.base_interval, + Duration::from_millis(100) + ); + assert_eq!( + shared.config.retry_backoff.max_interval, + Duration::from_millis(1000) + ); + assert_eq!( + shared.classifier.retry_on.as_ref(), + [tonic::Code::Unavailable] ); - let loaded = policy.load_config(); - assert_eq!(loaded.num_retries, 1); } #[test] - fn test_policy_update_config() { - let policy = GrpcRetryPolicy::default(); - assert_eq!(policy.load_config().num_retries, 1); + fn test_from_route_retry_unset_fields_use_defaults() { + // Only retry_on set: num_retries and backoff fall back to RetryConfig defaults. + let retry = RouteRetryConfig { + retry_on: "cancelled".into(), + num_retries: None, + base_interval: None, + max_interval: None, + }; + let shared = GrpcRetrySharedConfig::from_route_retry(&retry); + assert_eq!(shared.config.num_retries, 1); + assert_eq!(shared.config.retry_backoff, RetryBackoffConfig::default()); + assert_eq!( + shared.classifier.retry_on.as_ref(), + [tonic::Code::Cancelled] + ); + } - policy.update_config(RetryConfig::new().num_retries(3)); + // --- Per-request policy instantiation from a shared config (hot path) --- - let loaded = policy.load_config(); - assert_eq!(loaded.num_retries, 3); + #[test] + fn from_shared_instantiates_zeroed_state_sharing_config() { + // The hot path clones the shared-config `Arc` and zero-inits per-request + // state — no parsing or config copy. + let shared = Arc::new(GrpcRetrySharedConfig::from_route_retry(&RouteRetryConfig { + retry_on: "unavailable".into(), + num_retries: Some(2), + base_interval: None, + max_interval: None, + })); + let policy = RetryPolicy::from_shared(Arc::clone(&shared)); + + // Fresh per-request state. + assert_eq!(policy.attempts, 0); + assert!(policy.backoff.is_none()); + // The parsed config is shared by pointer, not copied. + assert!(Arc::ptr_eq(&policy.shared, &shared)); + assert_eq!(policy.shared.config.num_retries, 2); + + // A second policy from the same shared config points at the same config. + let policy2 = RetryPolicy::from_shared(Arc::clone(&shared)); + assert!(Arc::ptr_eq(&policy.shared, &policy2.shared)); } /// Verify that two concurrent requests using the same policy get independent @@ -685,9 +884,7 @@ mod tests { async fn test_retry_state_is_per_request() { let policy = GrpcRetryPolicy::new( RetryConfig::new().num_retries(2), - GrpcRetryClassifier { - retry_on: vec![tonic::Code::Unavailable], - }, + GrpcRetryClassifier::new(vec![tonic::Code::Unavailable]), ); // Simulate two independent request sessions by cloning the policy diff --git a/tonic-xds/src/client/route.rs b/tonic-xds/src/client/route.rs index 0e8ff99e9..b8b2d0272 100644 --- a/tonic-xds/src/client/route.rs +++ b/tonic-xds/src/client/route.rs @@ -22,6 +22,7 @@ * */ +use crate::client::retry::GrpcRetrySharedConfig; use crate::common::async_util::BoxFuture; use crate::xds::resource::route_config::{RouteConfigMetadata, RouteConfigResource}; use crate::xds::routing::RoutingError; @@ -50,6 +51,14 @@ pub(crate) struct RouteDecision { // Populated by the routing layer; consumed by the ring-hash picker (later PR). #[allow(dead_code)] pub request_hash: Option, + /// The matched route's shared retry config (RDS `RouteAction.retry_policy`), + /// or `None` when the route specifies no retry. Carried from the same config + /// snapshot the routing decision was made on, so the retry layer applies the + /// config for exactly the route this request took (gRFC A44). Already parsed + /// and validated at RDS-validation time and held behind an `Arc`, so the + /// retry layer instantiates a per-request policy from it with a pointer clone + /// (see `GrpcRetrySharedConfig`). + pub retry_config: Option>, } /// A hook that runs before xDS route selection. @@ -227,6 +236,7 @@ mod tests { Ok(RouteDecision { cluster: "test-cluster".to_string(), request_hash: None, + retry_config: None, }) } } @@ -301,6 +311,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("c".into()), + retry_config: None, }], }], metadata: RouteConfigMetadata::from_encoded( @@ -338,6 +349,7 @@ mod tests { Ok(RouteDecision { cluster: "c".to_string(), request_hash: None, + retry_config: None, }) } } @@ -442,6 +454,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster(cluster.into()), + retry_config: None, } } diff --git a/tonic-xds/src/xds/cache.rs b/tonic-xds/src/xds/cache.rs index 12ae472d5..ed99b3086 100644 --- a/tonic-xds/src/xds/cache.rs +++ b/tonic-xds/src/xds/cache.rs @@ -208,6 +208,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("cluster-1".to_string()), + retry_config: None, }], }], metadata: Default::default(), diff --git a/tonic-xds/src/xds/resource/route_config.rs b/tonic-xds/src/xds/resource/route_config.rs index 68a09b53e..9f896fd0a 100644 --- a/tonic-xds/src/xds/resource/route_config.rs +++ b/tonic-xds/src/xds/resource/route_config.rs @@ -25,11 +25,13 @@ //! Validated RouteConfiguration resource (RDS). use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::Duration; use bytes::Bytes; use envoy_types::pb::envoy::config::core::v3::Metadata; use envoy_types::pb::envoy::config::route::v3::{ - RouteConfiguration, RouteMatch, route, route_action, route_match, + RetryPolicy, RouteConfiguration, RouteMatch, route, route_action, route_match, }; use prost::Message; use regex::Regex; @@ -37,6 +39,7 @@ use xds_client::resource::TypeUrl; use xds_client::{Error, Resource}; use super::string_matcher::StringMatcher; +use crate::client::retry::GrpcRetrySharedConfig; /// A `typed_filter_metadata` entry — a `google.protobuf.Any` (a type URL plus an /// encoded message value). @@ -149,6 +152,53 @@ pub(crate) struct RouteConfigResource { pub metadata: RouteConfigMetadata, } +/// Raw Envoy retry settings parsed from a `RouteAction.retry_policy` or a +/// `VirtualHost.retry_policy` (RDS), used only as an intermediate carrier during +/// validation. Both proto fields are the same Envoy `RetryPolicy` message. +/// +/// This is the transport-neutral shape straight from the proto; +/// [`GrpcRetrySharedConfig::from_route_retry`] turns it into the validated, +/// gRPC-specific shared config that routes actually carry (see +/// [`RouteConfig::retry_config`]). +#[derive(Debug, Clone)] +pub(crate) struct RouteRetryConfig { + /// Envoy `retry_on` conditions, comma-separated (e.g. `"unavailable"`). + pub retry_on: String, + /// `num_retries` if set; `None` means the caller applies its own default. + pub num_retries: Option, + /// `retry_back_off.base_interval` if set. + pub base_interval: Option, + /// `retry_back_off.max_interval` if set. + pub max_interval: Option, +} + +impl RouteRetryConfig { + /// Parse an Envoy `RetryPolicy` proto into a [`RouteRetryConfig`]. + fn from_proto(rp: &RetryPolicy) -> Self { + let (base_interval, max_interval) = match rp.retry_back_off.as_ref() { + Some(backoff) => ( + backoff.base_interval.as_ref().and_then(proto_duration), + backoff.max_interval.as_ref().and_then(proto_duration), + ), + None => (None, None), + }; + Self { + retry_on: rp.retry_on.clone(), + num_retries: rp.num_retries.as_ref().map(|v| v.value), + base_interval, + max_interval, + } + } +} + +/// Convert a protobuf `Duration` to [`std::time::Duration`], returning `None` +/// for negative values (retry backoff intervals must be non-negative). +fn proto_duration(d: &envoy_types::pb::google::protobuf::Duration) -> Option { + let seconds = u64::try_from(d.seconds).ok()?; + let nanos = u32::try_from(d.nanos).ok()?; + Some(Duration::new(seconds, nanos)) +} + /// Validated virtual host with domain matching and routes. #[derive(Debug, Clone)] pub(crate) struct VirtualHostConfig { @@ -162,6 +212,27 @@ pub(crate) struct VirtualHostConfig { pub(crate) struct RouteConfig { pub match_criteria: RouteConfigMatch, pub action: RouteConfigAction, + /// This route's validated retry config, derived from its + /// `RouteAction.retry_policy` (RDS) when the `RouteConfiguration` is + /// validated, falling back to the enclosing `VirtualHost.retry_policy` when + /// the route sets none. `None` when neither specifies retry. + /// + /// Per gRFC A44 a route-level policy completely overrides the virtual host's + /// (values are not merged); routes that inherit the vhost policy share its + /// `Arc`. + /// + /// The `retry_on` conditions are parsed into gRPC status codes and defaults + /// applied **here, once at validation time** — not per request — so the hot + /// path never parses. Held behind an `Arc` so the routing layer can hand the + /// matched route's shared config to the retry layer (via `RouteDecision`) by + /// a cheap pointer clone, and the retry layer instantiates a per-request + /// policy from it with no allocation (see `GrpcRetrySharedConfig`). + /// + /// Per-route granularity: each request retries according to the exact route + /// it matched (gRFC A44). This holds a gRPC-specific config (rather than the + /// transport-neutral `RouteRetryConfig`) because deriving and validating it + /// once at RDS-validation time is what keeps the request path allocation-free. + pub retry_config: Option>, } /// Validated route match criteria. @@ -265,8 +336,17 @@ impl Resource for RouteConfigResource { } let mut routes = Vec::with_capacity(vh.routes.len()); + // gRFC A44: a `VirtualHost.retry_policy` applies to every route in the + // vhost that doesn't set its own; a route-level policy completely + // overrides it (values are not merged). Parse it once per vhost so all + // inheriting routes share the same `Arc`. + let vh_retry = vh.retry_policy.as_ref().map(|rp| { + Arc::new(GrpcRetrySharedConfig::from_route_retry( + &RouteRetryConfig::from_proto(rp), + )) + }); for route in vh.routes { - if let Some(validated_route) = validate_route(route)? { + if let Some(validated_route) = validate_route(route, vh_retry.as_ref())? { routes.push(validated_route); } } @@ -288,8 +368,13 @@ impl Resource for RouteConfigResource { /// Returns `Ok(None)` for routes that should be silently skipped (query param matchers, /// unsupported cluster specifiers like `cluster_header`). +/// +/// `vh_retry` is the enclosing virtual host's retry config (gRFC A44), if any. It is +/// used as the fallback for routes that don't set their own `RouteAction.retry_policy`; +/// a route-level policy takes precedence and completely overrides it. fn validate_route( route: envoy_types::pb::envoy::config::route::v3::Route, + vh_retry: Option<&Arc>, ) -> xds_client::Result> { let route_match = route .r#match @@ -306,11 +391,25 @@ fn validate_route( .action .ok_or_else(|| Error::Validation("route missing action field".into()))?; - let validated_action = match action { - route::Action::Route(route_action) => match validate_route_action(route_action)? { - Some(action) => action, - None => return Ok(None), - }, + let validated_action; + let retry_config; + match action { + route::Action::Route(route_action) => { + // Parse and validate the route's retry policy before `route_action` + // is consumed. Deriving the gRPC config here — once, at RDS-validation + // time — is what keeps the request hot path free of parsing and + // allocation (see `RouteConfig::retry_config`). A route-level policy + // takes precedence over the virtual host's (gRFC A44); the vhost + // fallback is applied below. + retry_config = route_action.retry_policy.as_ref().map(|rp| { + let raw = RouteRetryConfig::from_proto(rp); + Arc::new(GrpcRetrySharedConfig::from_route_retry(&raw)) + }); + match validate_route_action(route_action)? { + Some(action) => validated_action = action, + None => return Ok(None), + } + } // Per A28: action field must be "route", otherwise NACK. _ => { return Err(Error::Validation( @@ -322,6 +421,11 @@ fn validate_route( Ok(Some(RouteConfig { match_criteria, action: validated_action, + // gRFC A44: fall back to the virtual host's retry policy when the route + // has none. `retry_config` is `Some` iff the route set its own policy, so + // `or_else` implements route-over-vhost precedence (a complete override, + // not a field merge). Inheriting routes share the vhost's `Arc`. + retry_config: retry_config.or_else(|| vh_retry.cloned()), })) } @@ -498,8 +602,9 @@ impl RouteConfigResource { mod tests { use super::*; use envoy_types::pb::envoy::config::route::v3::{ - RouteAction, VirtualHost, route::Action, route_action::ClusterSpecifier, + RetryPolicy, RouteAction, VirtualHost, route::Action, route_action::ClusterSpecifier, }; + use envoy_types::pb::google::protobuf::UInt32Value; fn make_route(prefix: &str, cluster: &str) -> envoy_types::pb::envoy::config::route::v3::Route { envoy_types::pb::envoy::config::route::v3::Route { @@ -515,6 +620,33 @@ mod tests { } } + fn retry_policy(retry_on: &str, num_retries: u32) -> RetryPolicy { + RetryPolicy { + retry_on: retry_on.to_string(), + num_retries: Some(UInt32Value { value: num_retries }), + ..Default::default() + } + } + + fn make_route_with_retry( + prefix: &str, + cluster: &str, + retry: RetryPolicy, + ) -> envoy_types::pb::envoy::config::route::v3::Route { + envoy_types::pb::envoy::config::route::v3::Route { + r#match: Some(RouteMatch { + path_specifier: Some(route_match::PathSpecifier::Prefix(prefix.to_string())), + ..Default::default() + }), + action: Some(Action::Route(RouteAction { + cluster_specifier: Some(ClusterSpecifier::Cluster(cluster.to_string())), + retry_policy: Some(retry), + ..Default::default() + })), + ..Default::default() + } + } + fn make_route_config(name: &str) -> RouteConfiguration { RouteConfiguration { name: name.to_string(), @@ -977,4 +1109,55 @@ mod tests { assert_eq!(headers.len(), 1); assert_eq!(headers[0].name, "x-env"); } + + #[test] + fn test_vhost_retry_policy_inherited_and_route_overrides() { + // gRFC A44: a route with no policy inherits the virtual host's; a route + // with its own policy completely overrides it. Fields are private, so we + // verify the *selection* structurally via `Arc` identity — value mapping + // is covered by the `from_route_retry` tests in `client::retry`. + let rc = RouteConfiguration { + name: "rc".to_string(), + virtual_hosts: vec![VirtualHost { + name: "vh1".to_string(), + domains: vec!["*".to_string()], + retry_policy: Some(retry_policy("unavailable", 3)), + routes: vec![ + make_route_with_retry("/own", "c-own", retry_policy("cancelled", 2)), + make_route("/inherit-a", "c-a"), + make_route("/inherit-b", "c-b"), + ], + ..Default::default() + }], + ..Default::default() + }; + let validated = RouteConfigResource::validate(rc).unwrap(); + let routes = &validated.virtual_hosts[0].routes; + + let own = routes[0].retry_config.as_ref().expect("route-level policy"); + let inherit_a = routes[1].retry_config.as_ref().expect("inherited policy"); + let inherit_b = routes[2].retry_config.as_ref().expect("inherited policy"); + + // The overriding route gets its own config, distinct from the vhost's. + assert!(!Arc::ptr_eq(own, inherit_a)); + // Inheriting routes share the single vhost-level config `Arc`. + assert!(Arc::ptr_eq(inherit_a, inherit_b)); + } + + #[test] + fn test_no_retry_policy_yields_none() { + // No vhost policy and no route policy => the route carries no retry config. + let rc = RouteConfiguration { + name: "rc".to_string(), + virtual_hosts: vec![VirtualHost { + name: "vh1".to_string(), + domains: vec!["*".to_string()], + routes: vec![make_route("/", "c1")], + ..Default::default() + }], + ..Default::default() + }; + let validated = RouteConfigResource::validate(rc).unwrap(); + assert!(validated.virtual_hosts[0].routes[0].retry_config.is_none()); + } } diff --git a/tonic-xds/src/xds/resource_manager.rs b/tonic-xds/src/xds/resource_manager.rs index 9f0c80f1d..ea8c847b2 100644 --- a/tonic-xds/src/xds/resource_manager.rs +++ b/tonic-xds/src/xds/resource_manager.rs @@ -341,6 +341,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster((*c).into()), + retry_config: None, }) .collect(), }], @@ -366,6 +367,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster((*c).into()), + retry_config: None, }) .collect(), }], diff --git a/tonic-xds/src/xds/routing.rs b/tonic-xds/src/xds/routing.rs index 4740487fe..be7a9b415 100644 --- a/tonic-xds/src/xds/routing.rs +++ b/tonic-xds/src/xds/routing.rs @@ -142,14 +142,22 @@ fn resolve_route( .get(":path") .and_then(|v| v.to_str().ok()) .unwrap_or("/"); - let action = rc.route(authority, path, headers)?; - let cluster = match action { + let route = rc.matched_route(authority, path, headers)?; + let cluster = match &route.action { RouteConfigAction::Cluster(name) => name.clone(), RouteConfigAction::WeightedClusters(clusters) => select_weighted_cluster(clusters) .ok_or(RoutingError::EmptyWeightedClusters)? .to_string(), }; + // Carry the matched route's shared retry config (if any) so the retry layer + // applies the config for exactly the route this request took (gRFC A44). This + // is a cheap `Arc` clone of an already-parsed, validated config; the retry + // layer instantiates a per-request policy from it with no allocation. Because + // routing and retry read the same config snapshot, both act on the same RDS + // version (no cross-layer skew). + let retry_config = route.retry_config.clone(); + // gRFC A42 ring-hash request hash. The policy list is empty for now, so // `request_hash` resolves to `None` and the ring-hash picker falls back to a // random hash. @@ -161,6 +169,7 @@ fn resolve_route( Ok(RouteDecision { cluster, request_hash, + retry_config, }) } @@ -181,19 +190,37 @@ impl RouteConfigResource { /// Match a request and return the target cluster action. /// /// Performs domain matching on the authority, then walks routes in order - /// to find the first match. + /// to find the first match. Test-only convenience wrapper around + /// [`matched_route`](Self::matched_route); production routing uses + /// `matched_route` so it can also read the matched route's retry policy. + #[cfg(test)] pub(crate) fn route( &self, authority: &str, path: &str, headers: &http::HeaderMap, ) -> Result<&RouteConfigAction, RoutingError> { + self.matched_route(authority, path, headers) + .map(|route| &route.action) + } + + /// Match a request and return the full matched [`RouteConfig`]. + /// + /// Performs domain matching on the authority, then walks routes in order to + /// find the first match. Returns the whole matched route so callers can read + /// per-route fields (e.g. the retry policy) alongside the action. + pub(crate) fn matched_route( + &self, + authority: &str, + path: &str, + headers: &http::HeaderMap, + ) -> Result<&RouteConfig, RoutingError> { let vh = find_best_matching_virtual_host(authority, &self.virtual_hosts) .ok_or_else(|| RoutingError::NoMatchingVirtualHost(authority.to_string()))?; for route in &vh.routes { if route_matches(route, path, headers) { - return Ok(&route.action); + return Ok(route); } } @@ -391,6 +418,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster(cluster.into()), + retry_config: None, } } @@ -623,6 +651,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("c1".into()), + retry_config: None, }], }]); let headers = http::HeaderMap::new(); @@ -646,6 +675,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("c1".into()), + retry_config: None, }], }]); let headers = http::HeaderMap::new(); @@ -677,6 +707,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("cluster-prod".into()), + retry_config: None, }, simple_route("/", "cluster-default"), ], @@ -714,6 +745,7 @@ mod tests { weight: 30, }, ]), + retry_config: None, }], }]); let action = rc.route("host", "/", &http::HeaderMap::new()).unwrap(); @@ -734,6 +766,7 @@ mod tests { match_fraction: Some(0), }, action: RouteConfigAction::Cluster("never".into()), + retry_config: None, }, simple_route("/", "fallback"), ], @@ -757,6 +790,7 @@ mod tests { match_fraction: Some(1_000_000), }, action: RouteConfigAction::Cluster("always".into()), + retry_config: None, }], }]); for _ in 0..100 { @@ -786,6 +820,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("versioned".into()), + retry_config: None, }, simple_route("/", "default"), ], @@ -881,6 +916,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("grpc".into()), + retry_config: None, }, simple_route("/", "fallback"), ], @@ -916,6 +952,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("matched".into()), + retry_config: None, }], }]); @@ -953,6 +990,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("matched".into()), + retry_config: None, }, simple_route("/", "fallback"), ], @@ -1011,6 +1049,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("matched".into()), + retry_config: None, }, simple_route("/", "fallback"), ], @@ -1052,6 +1091,7 @@ mod tests { match_fraction: None, }, action: RouteConfigAction::Cluster("matched".into()), + retry_config: None, }, simple_route("/", "fallback"), ], From d4225428fbdf5dee4d0e309da3f2a8fe4c1df12b Mon Sep 17 00:00:00 2001 From: Yu Liu <60283975+LYZJU2019@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:16:06 -0700 Subject: [PATCH 2/4] refactor(tonic-xds): address PR review for RDS-driven retry Compile the gRPC retry config once per RDS update in a new RoutingSnapshot (client layer), keeping the xDS resource types free of gRPC business logic; the resource layer now carries only the validated RouteRetryConfig. Routing and retry read one bundled snapshot, so the request hot path stays a map lookup plus an Arc clone. - Reject out-of-range google.protobuf.Duration values and guard backoff max-interval math with checked_mul (gRFC A44). - Skip retry parsing for routes that are dropped during validation. - Treat an empty parsed retry_on set as "no policy" (return None) so it does not mask connection-level retries. - RetryLayer::new takes the shared config Arc directly. --- tonic-xds/src/client/channel.rs | 26 +- tonic-xds/src/client/circuit_breaking.rs | 2 +- tonic-xds/src/client/retry.rs | 216 +++++++---------- tonic-xds/src/client/route.rs | 93 ++++++-- tonic-xds/src/xds/resource/route_config.rs | 262 +++++++++++++++------ tonic-xds/src/xds/routing.rs | 26 +- 6 files changed, 380 insertions(+), 245 deletions(-) diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index ea1133895..5585bc24a 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -46,7 +46,9 @@ use xds_client::{ ClientConfig, MetricsRecorder, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient, }; -use crate::client::retry::{GrpcRetryPolicy, RetryLayer}; +#[cfg(test)] +use crate::client::retry::GrpcRetryPolicy; +use crate::client::retry::{GrpcRetrySharedConfig, RetryLayer}; /// Configuration for building [`XdsChannel`] / [`XdsChannelGrpc`]. #[derive(Clone, Debug)] @@ -368,16 +370,14 @@ impl XdsChannelBuilder { ) -> XdsChannelGrpc { let router: Arc = Arc::new(XdsRouter::new(&cache)); - // Retry config is control-plane-driven from RDS, per route. It is parsed - // and validated once, when the `RouteConfiguration` is validated. The - // routing layer (outer) stamps the matched route's shared retry config - // into the request's `RouteDecision`; the retry layer reads the shared - // config `Arc` and instantiates a per-request policy from it, so the - // request hot path does no parsing or allocation. The default below is - // the fallback used when a request carries no route retry config (non-xDS - // callers, or a route with no retry policy). See - // [`RetryLayer`](crate::client::retry::RetryLayer). - let retry_layer = RetryLayer::new(GrpcRetryPolicy::default()); + // Retry config is control-plane-driven from RDS, per route, compiled once + // per RDS update in the routing layer and carried on the request's + // `RouteDecision`; the retry layer reads that shared config `Arc` and + // instantiates a per-request policy from it, so the request hot path does + // no parsing or allocation. The default below is the fallback used when a + // request carries no route retry config (non-xDS callers, or a route with + // no retry policy). See [`RetryLayer`](crate::client::retry::RetryLayer). + let retry_layer = RetryLayer::new(Arc::new(GrpcRetrySharedConfig::default())); #[cfg(feature = "_tls-any")] let discovery: Arc< @@ -432,7 +432,7 @@ impl XdsChannelBuilder { interceptor: Option>, ) -> XdsChannelGrpc { let routing_layer = XdsRoutingLayer::new(router, interceptor, self.authority()); - let retry_layer = RetryLayer::new(retry_policy); + let retry_layer = RetryLayer::new(retry_policy.into_shared()); let cluster_registry = Arc::new(ClusterClientRegistryGrpc::new()); let lb_service = XdsLbService::new(cluster_registry, discovery); let inner = ServiceBuilder::new() @@ -532,7 +532,7 @@ mod tests { fn route( &self, _input: &RouteInput<'_>, - _config: &crate::xds::resource::route_config::RouteConfigResource, + _config: &crate::client::route::RoutingSnapshot, ) -> Result { Ok(RouteDecision { cluster: "test-cluster".to_string(), diff --git a/tonic-xds/src/client/circuit_breaking.rs b/tonic-xds/src/client/circuit_breaking.rs index 3b5a60029..a5380270e 100644 --- a/tonic-xds/src/client/circuit_breaking.rs +++ b/tonic-xds/src/client/circuit_breaking.rs @@ -997,7 +997,7 @@ mod tests { }, ); let mut service = tower::ServiceBuilder::new() - .layer(RetryLayer::new(policy)) + .layer(RetryLayer::new(policy.into_shared())) .layer(CircuitBreakingLayer::new(breakers.clone())) .service(service); diff --git a/tonic-xds/src/client/retry.rs b/tonic-xds/src/client/retry.rs index 8e718ee07..74ade7a1d 100644 --- a/tonic-xds/src/client/retry.rs +++ b/tonic-xds/src/client/retry.rs @@ -129,7 +129,9 @@ impl RetryBackoffConfig { pub(crate) fn new(base_interval: Duration) -> Self { let base_interval = base_interval.max(MIN_BACKOFF); Self { - max_interval: base_interval * 10, + // `checked_mul` guards the (already range-checked) `base_interval` + // against overflow; the fallback keeps `max_interval >= base`. + max_interval: base_interval.checked_mul(10).unwrap_or(base_interval), base_interval, backoff_multiplier: 2.0, } @@ -206,14 +208,9 @@ impl Default for RetryConfig { /// Default gRPC [`RetryClassifier`]: retries on a retryable connection error or a /// retryable gRPC status code, and stamps the `grpc-previous-rpc-attempts` header /// on each retry per the gRPC spec. -/// -/// The retryable gRPC status codes are parsed once per `RouteConfiguration` and -/// held behind an `Arc<[Code]>`, so cloning a classifier (per request, or on each -/// tower retry) is a pointer bump rather than a re-parse or re-allocation. A new -/// set is built only when RDS changes; see [`RetryLayer`]. #[derive(Debug, Clone)] pub(crate) struct GrpcRetryClassifier { - /// gRPC status codes that should be retried. Shared (parsed once per config). + /// gRPC status codes that should be retried. retry_on: Arc<[tonic::Code]>, } @@ -266,27 +263,24 @@ fn make_backoff(config: &RetryBackoffConfig) -> backoff::ExponentialBackoff { .build() } -/// Immutable, shared retry configuration — the transport-agnostic knobs -/// (attempt cap, backoff) plus the transport-specific classifier `C`. Built and -/// validated **once** when a `RouteConfiguration` is validated (see -/// [`GrpcRetrySharedConfig::from_route_retry`]) and shared across every request -/// that matches the route via an [`Arc`], so per-request setup never re-parses -/// or re-allocates. +/// Immutable, shared retry configuration: the transport-agnostic knobs (attempt +/// cap, backoff) plus the transport-specific classifier `C`. Built once when a +/// `RouteConfiguration` is validated (see +/// [`GrpcRetrySharedConfig::from_route_retry`]) and shared across matching +/// requests via an [`Arc`]. /// -/// Kept separate from the per-request retry *state* ([`RetryPolicy`]) so that -/// instantiating a policy for a request is just an `Arc` pointer clone plus a -/// zero-field state init — see [`RetryPolicy::from_shared`]. +/// Kept separate from the per-request retry state ([`RetryPolicy`]) so that +/// instantiating a policy is an `Arc` clone plus a zero-field init. #[derive(Debug)] pub(crate) struct RetrySharedConfig { - /// Attempt cap and backoff schedule, shared by every [`RetryClassifier`]. + /// Attempt cap and backoff schedule. config: RetryConfig, /// Decides retryability and per-retry request mutation for the transport. classifier: C, } impl RetrySharedConfig { - /// Create a shared retry config from an attempt/backoff [`RetryConfig`] and a - /// transport classifier. + /// Create a shared retry config from a [`RetryConfig`] and a classifier. pub(crate) fn new(config: RetryConfig, classifier: C) -> Self { Self { config, classifier } } @@ -298,23 +292,14 @@ impl Default for RetrySharedConfig { } } -/// Per-request retry *state* — a pointer to the shared, immutable +/// Per-request retry *state*: a pointer to the shared, immutable /// [`RetrySharedConfig`] plus the mutable state for one request (backoff cursor -/// and attempt counter). Transport-specific decisions live in the classifier `C` -/// (see [`RetryClassifier`]). -/// -/// Instantiating one for a request ([`RetryPolicy::from_shared`]) is a hot-path -/// operation: it clones a single `Arc` (a pointer bump) and zero-inits the state -/// fields — no parsing, allocation, or config copying. -/// -/// Implements [`tower::retry::Policy`]. Tower's `Retry` service clones the policy -/// for each request, so `backoff` and `attempts` track per-request retry state -/// while the shared config stays behind the `Arc`. The retry *state machine* -/// stays entirely in tower's `Retry`/`ResponseFuture`; this type only implements -/// the `Policy` trait, so no state machine is reimplemented here. +/// and attempt counter). Implements [`tower::retry::Policy`]; tower's `Retry` +/// clones it per request, so the state is per-request while the config stays +/// shared behind the `Arc`. #[derive(Clone, Debug)] pub(crate) struct RetryPolicy { - /// Immutable config shared across all requests on this route (pointer clone). + /// Immutable config shared across all requests on this route. shared: Arc>, /// Backoff state for the current request, created from config on first retry. backoff: Option, @@ -323,17 +308,14 @@ pub(crate) struct RetryPolicy { } impl RetryPolicy { - /// Create a policy from a shared config and a classifier, allocating a fresh - /// [`Arc`] for the shared config. Prefer [`from_shared`](Self::from_shared) - /// on the hot path, where the shared config already lives behind an `Arc`. + /// Create a policy from a config and classifier, allocating the shared `Arc`. + /// Prefer [`from_shared`](Self::from_shared) on the hot path. pub(crate) fn new(config: RetryConfig, classifier: C) -> Self { Self::from_shared(Arc::new(RetrySharedConfig::new(config, classifier))) } - /// Instantiate per-request retry state from an already-shared config. - /// - /// Hot path: a single `Arc` pointer clone (the caller's) plus a zero-field - /// state init. No parsing, allocation, or config copy. + /// Instantiate per-request state from an already-shared config: an `Arc` + /// pointer clone plus a zero-field init. This is the hot path. pub(crate) fn from_shared(shared: Arc>) -> Self { Self { shared, @@ -342,11 +324,14 @@ impl RetryPolicy { } } - /// Get or create the backoff, and advance it to the next delay. - /// - /// Only called on the cold path (when a request is actually being retried), - /// never for successful requests. Borrows `shared` and `backoff` disjointly, - /// so it reads the shared backoff config by reference without cloning. + /// Consume the policy and return its shared config, discarding the + /// per-request state. + pub(crate) fn into_shared(self) -> Arc> { + self.shared + } + + /// Get or lazily create the backoff and advance it to the next delay. Only + /// reached when a request is actually being retried. fn backoff_next(&mut self) -> Duration { let backoff_config = &self.shared.config.retry_backoff; let backoff = self @@ -366,16 +351,17 @@ impl Default for RetryPolicy { impl RetrySharedConfig { /// Build a shared gRPC retry config from a route's [`RouteRetryConfig`] (RDS - /// `RouteAction.retry_policy`). Unset Envoy fields fall back to - /// [`RetryConfig`] defaults. + /// `RouteAction.retry_policy`), parsing `retry_on` into [`tonic::Code`]s. + /// Unset Envoy fields fall back to [`RetryConfig`] defaults. /// - /// This is where `retry_on` is parsed into [`tonic::Code`]s and the - /// retryable-code set is allocated, so it runs **once, when the - /// `RouteConfiguration` is validated** — not per request. The result is - /// wrapped in an [`Arc`] and carried on the matched route, so instantiating a - /// per-request policy is just a pointer clone (see - /// [`RetryPolicy::from_shared`]). - pub(crate) fn from_route_retry(retry: &RouteRetryConfig) -> Self { + /// Returns `None` when no `retry_on` condition maps to a gRPC status code + /// (gRFC A44): an empty set means "no retry policy" for the route, so it + /// falls back to the layer default rather than masking connection retries. + pub(crate) fn from_route_retry(retry: &RouteRetryConfig) -> Option { + let retry_on = grpc_retry_on_codes(&retry.retry_on); + if retry_on.is_empty() { + return None; + } let mut config = RetryConfig::new(); if let Some(num_retries) = retry.num_retries { config = config.num_retries(num_retries); @@ -387,10 +373,7 @@ impl RetrySharedConfig { } config = config.retry_backoff(backoff); } - Self::new( - config, - GrpcRetryClassifier::new(grpc_retry_on_codes(&retry.retry_on)), - ) + Some(Self::new(config, GrpcRetryClassifier::new(retry_on))) } } @@ -453,61 +436,34 @@ where /// Non-breaking alias: existing gRPC callers keep the same name and behavior. pub(crate) type GrpcRetryPolicy = RetryPolicy; -/// Shared, immutable gRPC retry config (attempt cap, backoff, and the retryable -/// [`tonic::Code`] set). Built and validated once per `RouteConfiguration` (see -/// [`RetrySharedConfig::from_route_retry`]) and carried behind an [`Arc`] on the -/// matched route, so instantiating a per-request [`GrpcRetryPolicy`] is a pointer -/// clone (see [`RetryPolicy::from_shared`]). +/// Shared, immutable gRPC retry config (attempt cap, backoff, retryable +/// [`tonic::Code`] set); the [`GrpcRetryClassifier`] specialization of +/// [`RetrySharedConfig`]. pub(crate) type GrpcRetrySharedConfig = RetrySharedConfig; /// Tower [`Layer`] that wraps a gRPC service with retry support. /// -/// Converts the request body into a [`SharedBody`] (cloneable) and constructs -/// a fresh [`tower::retry::Retry`] service per request so that each request -/// gets its own retry state. -/// -/// # Per-route, control-plane-driven policy -/// -/// The active retry config is selected **per request** from the route the -/// request matched. The routing layer (which runs immediately outside this one) -/// stamps the matched route's shared retry config into the request's -/// [`RouteDecision`], taken from the same config snapshot it routed with. Because -/// both layers read one snapshot, routing and retry always act on the same RDS -/// version — there is no cross-layer version skew — and each request retries -/// according to the exact route it took (gRFC A44). -/// +/// Converts the request body into a cloneable [`SharedBody`] and builds a fresh +/// [`tower::retry::Retry`] per request. The retry config is selected per request +/// from the route the request matched: the routing layer (just outside this one) +/// stamps the matched route's shared config into the request's [`RouteDecision`]. /// Requests with no [`RouteDecision`] (non-xDS callers) or whose matched route /// carries no retry policy use `fallback`. /// -/// # Hot path -/// -/// The gRPC retry config (`retry_on` codes, attempt cap, backoff) is parsed and -/// validated **once**, when the `RouteConfiguration` is validated, and shared -/// behind an [`Arc`] on the matched route (see [`GrpcRetrySharedConfig`]). This -/// layer never parses or derives on the request path: it reads the shared config -/// `Arc` from the [`RouteDecision`], clones the pointer, and instantiates a -/// [`GrpcRetryPolicy`] with fresh per-request state (see -/// [`RetryPolicy::from_shared`]). No parsing, allocation, or locking. -/// -/// This layer is gRPC-specific because it reads the concrete [`RouteDecision`] -/// extension; the retry *engine* ([`RetryPolicy`], [`RetrySharedConfig`], and the -/// [`RetryClassifier`] seam) stays transport-agnostic. +/// gRPC-specific because it reads the concrete [`RouteDecision`] extension; the +/// retry engine ([`RetryPolicy`], [`RetrySharedConfig`], [`RetryClassifier`]) +/// stays transport-agnostic. #[derive(Clone)] pub(crate) struct RetryLayer { - /// Shared config used when a request carries no per-route retry config - /// (non-xDS callers, or a matched route with no `RouteAction.retry_policy`). + /// Config used when a request carries no per-route retry config. fallback: Arc, } impl RetryLayer { - /// Create a retry layer whose `fallback` policy is used when a request - /// carries no per-route retry config. Only the policy's shared config is - /// kept; its per-request state is discarded. Route-specific configs are read - /// from the request's [`RouteDecision`] (see [`RetryLayer`]). - pub(crate) fn new(fallback: GrpcRetryPolicy) -> Self { - Self { - fallback: fallback.shared, - } + /// Create a layer with the given `fallback` shared config, used when a + /// request carries no per-route retry config (see [`RetryLayer`]). + pub(crate) fn new(fallback: Arc) -> Self { + Self { fallback } } } @@ -523,15 +479,12 @@ impl Layer for RetryLayer { } /// Service that converts request bodies to [`SharedBody`] and retries via -/// [`tower::retry::Retry`]. The retry config is selected per request from the -/// route the request matched (via [`RouteDecision`]) and is already parsed and -/// shared behind an [`Arc`], so the hot path performs no parsing or allocation -/// (see [`RetryLayer`]). +/// [`tower::retry::Retry`], selecting the per-request config from the matched +/// route's [`RouteDecision`] (see [`RetryLayer`]). #[derive(Clone)] pub(crate) struct RetryService { inner: S, - /// Shared config used when a request carries no per-route retry config, - /// shared (via `Arc`) with the layer and every per-request clone. + /// Config used when a request carries no per-route retry config. fallback: Arc, } @@ -559,13 +512,8 @@ where } fn call(&mut self, request: Request) -> Self::Future { - // Select the shared retry config for the route this request matched. The - // routing layer (just outside this one) stamped the matched route's shared - // config into the `RouteDecision` from the same config snapshot it routed - // with. The config is already parsed and validated (done once when the - // `RouteConfiguration` was validated), so this is a pointer clone plus a - // zero-field state init — no parsing or allocation. Fall back when the - // request carries no route retry config. + // Use the shared config the routing layer stamped into the RouteDecision + // for the route this request matched, falling back when it carries none. let shared = request .extensions() .get::() @@ -807,7 +755,7 @@ mod tests { assert_eq!(config.retry_backoff, backoff); } - // --- Building a policy from a route's retry config --- + // --- from_route_retry tests --- #[test] fn test_from_route_retry_maps_fields() { @@ -817,7 +765,7 @@ mod tests { base_interval: Some(Duration::from_millis(100)), max_interval: Some(Duration::from_millis(1000)), }; - let shared = GrpcRetrySharedConfig::from_route_retry(&retry); + let shared = GrpcRetrySharedConfig::from_route_retry(&retry).expect("codes present"); assert_eq!(shared.config.num_retries, 3); assert_eq!( shared.config.retry_backoff.base_interval, @@ -835,14 +783,13 @@ mod tests { #[test] fn test_from_route_retry_unset_fields_use_defaults() { - // Only retry_on set: num_retries and backoff fall back to RetryConfig defaults. let retry = RouteRetryConfig { retry_on: "cancelled".into(), num_retries: None, base_interval: None, max_interval: None, }; - let shared = GrpcRetrySharedConfig::from_route_retry(&retry); + let shared = GrpcRetrySharedConfig::from_route_retry(&retry).expect("codes present"); assert_eq!(shared.config.num_retries, 1); assert_eq!(shared.config.retry_backoff, RetryBackoffConfig::default()); assert_eq!( @@ -851,28 +798,39 @@ mod tests { ); } - // --- Per-request policy instantiation from a shared config (hot path) --- - #[test] - fn from_shared_instantiates_zeroed_state_sharing_config() { - // The hot path clones the shared-config `Arc` and zero-inits per-request - // state — no parsing or config copy. - let shared = Arc::new(GrpcRetrySharedConfig::from_route_retry(&RouteRetryConfig { - retry_on: "unavailable".into(), - num_retries: Some(2), + fn test_from_route_retry_empty_codes_yields_none() { + // `retry_on` with only non-gRPC tokens maps to no gRPC codes, so no + // policy is produced and connection retries are not masked (gRFC A44). + let retry = RouteRetryConfig { + retry_on: "5xx,reset".into(), + num_retries: Some(3), base_interval: None, max_interval: None, - })); + }; + assert!(GrpcRetrySharedConfig::from_route_retry(&retry).is_none()); + } + + // --- from_shared tests --- + + #[test] + fn from_shared_instantiates_zeroed_state_sharing_config() { + let shared = Arc::new( + GrpcRetrySharedConfig::from_route_retry(&RouteRetryConfig { + retry_on: "unavailable".into(), + num_retries: Some(2), + base_interval: None, + max_interval: None, + }) + .expect("codes present"), + ); let policy = RetryPolicy::from_shared(Arc::clone(&shared)); - // Fresh per-request state. assert_eq!(policy.attempts, 0); assert!(policy.backoff.is_none()); - // The parsed config is shared by pointer, not copied. assert!(Arc::ptr_eq(&policy.shared, &shared)); assert_eq!(policy.shared.config.num_retries, 2); - // A second policy from the same shared config points at the same config. let policy2 = RetryPolicy::from_shared(Arc::clone(&shared)); assert!(Arc::ptr_eq(&policy.shared, &policy2.shared)); } diff --git a/tonic-xds/src/client/route.rs b/tonic-xds/src/client/route.rs index b8b2d0272..e9e95a922 100644 --- a/tonic-xds/src/client/route.rs +++ b/tonic-xds/src/client/route.rs @@ -24,9 +24,13 @@ use crate::client::retry::GrpcRetrySharedConfig; use crate::common::async_util::BoxFuture; -use crate::xds::resource::route_config::{RouteConfigMetadata, RouteConfigResource}; +use crate::xds::resource::route_config::{ + RouteConfigMetadata, RouteConfigResource, RouteRetryConfig, +}; use crate::xds::routing::RoutingError; use http::Request; +use std::collections::HashMap; +use std::ops::Deref; use std::sync::Arc; use std::task::{Context, Poll}; use tower::{BoxError, Layer, Service}; @@ -51,13 +55,12 @@ pub(crate) struct RouteDecision { // Populated by the routing layer; consumed by the ring-hash picker (later PR). #[allow(dead_code)] pub request_hash: Option, - /// The matched route's shared retry config (RDS `RouteAction.retry_policy`), - /// or `None` when the route specifies no retry. Carried from the same config - /// snapshot the routing decision was made on, so the retry layer applies the - /// config for exactly the route this request took (gRFC A44). Already parsed - /// and validated at RDS-validation time and held behind an `Arc`, so the - /// retry layer instantiates a per-request policy from it with a pointer clone - /// (see `GrpcRetrySharedConfig`). + /// The matched route's compiled retry config (RDS `RouteAction.retry_policy`), + /// or `None` when the route specifies no retry. Resolved from the same + /// [`RoutingSnapshot`] the routing decision was made on, so the retry layer + /// applies the config for exactly the route this request took (gRFC A44). + /// Compiled once per RDS update and held behind an `Arc`, so the retry layer + /// instantiates a per-request policy from it with a pointer clone. pub retry_config: Option>, } @@ -73,6 +76,62 @@ pub trait PreRouteInterceptor: Send + Sync + 'static { fn on_request(&self, headers: &mut http::HeaderMap, metadata: &RouteConfigMetadata); } +/// A per-RDS-update routing snapshot: the validated [`RouteConfigResource`] +/// bundled with the gRPC retry configs compiled from it, once, when the snapshot +/// is installed. +/// +/// Holding both in one `Arc` lets the routing layer resolve a route and its +/// retry config from a single, consistent RDS version, and keeps the request hot +/// path to a map lookup plus an `Arc` clone (no parsing or allocation). +/// +/// Compiling the gRPC config here, rather than in the xDS resource layer, keeps +/// the resource types free of gRPC business logic. +#[derive(Debug, Default)] +pub(crate) struct RoutingSnapshot { + resource: Arc, + /// Compiled retry config per route, keyed by the address of the route's + /// [`RouteRetryConfig`] `Arc`. Routes that share one config (vhost + /// inheritance) resolve to a single entry; routes whose `retry_on` maps to no + /// gRPC code have no entry (gRFC A44). + retry: HashMap>, +} + +impl RoutingSnapshot { + /// Builds a snapshot from a validated resource, compiling each distinct + /// route retry config once (gRFC A44). + pub(crate) fn new(resource: Arc) -> Self { + let mut retry: HashMap> = HashMap::new(); + for vhost in &resource.virtual_hosts { + for route in &vhost.routes { + if let Some(config) = &route.retry_config { + let key = Arc::as_ptr(config) as usize; + if let std::collections::hash_map::Entry::Vacant(entry) = retry.entry(key) + && let Some(shared) = GrpcRetrySharedConfig::from_route_retry(config) + { + entry.insert(Arc::new(shared)); + } + } + } + } + Self { resource, retry } + } + + /// The compiled gRPC retry config for a route's [`RouteRetryConfig`], if any. + pub(crate) fn retry_for( + &self, + config: Option<&Arc>, + ) -> Option> { + config.and_then(|config| self.retry.get(&(Arc::as_ptr(config) as usize)).cloned()) + } +} + +impl Deref for RoutingSnapshot { + type Target = RouteConfigResource; + fn deref(&self) -> &Self::Target { + &self.resource + } +} + /// A route config obtained in one step, to serve a single request with. /// /// Two cases so the common one -- a config is already in effect -- costs @@ -80,15 +139,15 @@ pub trait PreRouteInterceptor: Send + Sync + 'static { /// wait for the first config. pub(crate) enum AcquiredConfig { /// A config is already in effect. - Ready(Arc), + Ready(Arc), /// None has arrived yet; await this for the first one, bounded by an /// implementation-defined timeout. - Pending(BoxFuture, RoutingError>>), + Pending(BoxFuture, RoutingError>>), } impl AcquiredConfig { /// Resolves to the config, awaiting only when one is not already available. - pub(crate) async fn get(self) -> Result, RoutingError> { + pub(crate) async fn get(self) -> Result, RoutingError> { match self { Self::Ready(config) => Ok(config), Self::Pending(wait) => wait.await, @@ -109,7 +168,7 @@ pub(crate) trait Router: Send + Sync + 'static { fn route( &self, input: &RouteInput<'_>, - config: &RouteConfigResource, + config: &RoutingSnapshot, ) -> Result; } @@ -224,13 +283,13 @@ mod tests { impl Router for CaptureAuthorityRouter { fn acquire(&self) -> AcquiredConfig { - AcquiredConfig::Ready(Arc::new(RouteConfigResource::default())) + AcquiredConfig::Ready(Arc::new(RoutingSnapshot::default())) } fn route( &self, input: &RouteInput<'_>, - _config: &RouteConfigResource, + _config: &RoutingSnapshot, ) -> Result { *self.captured.lock().unwrap() = Some(input.authority.to_string()); Ok(RouteDecision { @@ -333,7 +392,7 @@ mod tests { #[tokio::test] async fn releases_the_route_config_before_calling_the_inner_service() { struct SharedConfigRouter { - config: Arc, + config: Arc, } impl Router for SharedConfigRouter { @@ -344,7 +403,7 @@ mod tests { fn route( &self, _input: &RouteInput<'_>, - _config: &RouteConfigResource, + _config: &RoutingSnapshot, ) -> Result { Ok(RouteDecision { cluster: "c".to_string(), @@ -354,7 +413,7 @@ mod tests { } } - let config = Arc::new(RouteConfigResource::default()); + let config = Arc::new(RoutingSnapshot::default()); let router: Arc = Arc::new(SharedConfigRouter { config: config.clone(), }); diff --git a/tonic-xds/src/xds/resource/route_config.rs b/tonic-xds/src/xds/resource/route_config.rs index 9f896fd0a..f429c6343 100644 --- a/tonic-xds/src/xds/resource/route_config.rs +++ b/tonic-xds/src/xds/resource/route_config.rs @@ -39,7 +39,6 @@ use xds_client::resource::TypeUrl; use xds_client::{Error, Resource}; use super::string_matcher::StringMatcher; -use crate::client::retry::GrpcRetrySharedConfig; /// A `typed_filter_metadata` entry — a `google.protobuf.Any` (a type URL plus an /// encoded message value). @@ -152,14 +151,9 @@ pub(crate) struct RouteConfigResource { pub metadata: RouteConfigMetadata, } -/// Raw Envoy retry settings parsed from a `RouteAction.retry_policy` or a -/// `VirtualHost.retry_policy` (RDS), used only as an intermediate carrier during -/// validation. Both proto fields are the same Envoy `RetryPolicy` message. -/// -/// This is the transport-neutral shape straight from the proto; -/// [`GrpcRetrySharedConfig::from_route_retry`] turns it into the validated, -/// gRPC-specific shared config that routes actually carry (see -/// [`RouteConfig::retry_config`]). +/// Validated Envoy retry settings (gRFC A44) parsed from a `RouteAction` or +/// `VirtualHost` `retry_policy` (RDS). A resource-layer type; the routing layer +/// compiles it into the gRPC retry config once per RDS update. #[derive(Debug, Clone)] pub(crate) struct RouteRetryConfig { /// Envoy `retry_on` conditions, comma-separated (e.g. `"unavailable"`). @@ -173,32 +167,84 @@ pub(crate) struct RouteRetryConfig { } impl RouteRetryConfig { - /// Parse an Envoy `RetryPolicy` proto into a [`RouteRetryConfig`]. - fn from_proto(rp: &RetryPolicy) -> Self { + /// Parse and validate an Envoy `RetryPolicy` (gRFC A44). Returns + /// `Err(Validation)` — so the xDS client NACKs — when `num_retries < 1`, or + /// when `retry_back_off` is set with a `base_interval` or `max_interval` + /// that is not greater than zero. + fn from_proto(rp: &RetryPolicy) -> xds_client::Result { + let num_retries = match rp.num_retries.as_ref().map(|v| v.value) { + Some(0) => { + return Err(Error::Validation( + "retry_policy.num_retries must be >= 1".into(), + )); + } + other => other, + }; + let (base_interval, max_interval) = match rp.retry_back_off.as_ref() { - Some(backoff) => ( - backoff.base_interval.as_ref().and_then(proto_duration), - backoff.max_interval.as_ref().and_then(proto_duration), - ), + Some(backoff) => { + let base_interval = backoff + .base_interval + .as_ref() + .and_then(proto_duration) + .filter(|d| !d.is_zero()) + .ok_or_else(|| { + Error::Validation( + "retry_policy.retry_back_off.base_interval must be greater than 0" + .into(), + ) + })?; + let max_interval = backoff + .max_interval + .as_ref() + .map(|m| { + proto_duration(m).filter(|d| !d.is_zero()).ok_or_else(|| { + Error::Validation( + "retry_policy.retry_back_off.max_interval must be greater than 0" + .into(), + ) + }) + }) + .transpose()?; + (Some(base_interval), max_interval) + } None => (None, None), }; - Self { + + Ok(Self { retry_on: rp.retry_on.clone(), - num_retries: rp.num_retries.as_ref().map(|v| v.value), + num_retries, base_interval, max_interval, - } + }) } } +/// Maximum `seconds` for a well-formed `google.protobuf.Duration`. +const MAX_PROTO_DURATION_SECONDS: i64 = 315_576_000_000; + /// Convert a protobuf `Duration` to [`std::time::Duration`], returning `None` -/// for negative values (retry backoff intervals must be non-negative). +/// for values outside the documented `google.protobuf.Duration` range +/// (negatives included). Out-of-range values are rejected here so an invalid +/// retry policy fails validation rather than overflowing later backoff math. fn proto_duration(d: &envoy_types::pb::google::protobuf::Duration) -> Option { + if !(0..=MAX_PROTO_DURATION_SECONDS).contains(&d.seconds) + || !(0..=999_999_999).contains(&d.nanos) + { + return None; + } let seconds = u64::try_from(d.seconds).ok()?; let nanos = u32::try_from(d.nanos).ok()?; Some(Duration::new(seconds, nanos)) } +/// Parse and validate (gRFC A44) an Envoy `RetryPolicy` into a shared, +/// resource-layer [`RouteRetryConfig`]. Routes that inherit a virtual host's +/// policy share one `Arc`. +fn parse_retry(rp: &RetryPolicy) -> xds_client::Result> { + Ok(Arc::new(RouteRetryConfig::from_proto(rp)?)) +} + /// Validated virtual host with domain matching and routes. #[derive(Debug, Clone)] pub(crate) struct VirtualHostConfig { @@ -212,27 +258,12 @@ pub(crate) struct VirtualHostConfig { pub(crate) struct RouteConfig { pub match_criteria: RouteConfigMatch, pub action: RouteConfigAction, - /// This route's validated retry config, derived from its - /// `RouteAction.retry_policy` (RDS) when the `RouteConfiguration` is - /// validated, falling back to the enclosing `VirtualHost.retry_policy` when - /// the route sets none. `None` when neither specifies retry. - /// - /// Per gRFC A44 a route-level policy completely overrides the virtual host's - /// (values are not merged); routes that inherit the vhost policy share its - /// `Arc`. - /// - /// The `retry_on` conditions are parsed into gRPC status codes and defaults - /// applied **here, once at validation time** — not per request — so the hot - /// path never parses. Held behind an `Arc` so the routing layer can hand the - /// matched route's shared config to the retry layer (via `RouteDecision`) by - /// a cheap pointer clone, and the retry layer instantiates a per-request - /// policy from it with no allocation (see `GrpcRetrySharedConfig`). - /// - /// Per-route granularity: each request retries according to the exact route - /// it matched (gRFC A44). This holds a gRPC-specific config (rather than the - /// transport-neutral `RouteRetryConfig`) because deriving and validating it - /// once at RDS-validation time is what keeps the request path allocation-free. - pub retry_config: Option>, + /// Validated retry settings for this route (gRFC A44), or `None` when + /// neither the route nor its virtual host sets a policy. A route-level + /// policy completely overrides the virtual host's (values are not merged); + /// routes that inherit the vhost policy share one `Arc`. The routing layer + /// compiles this into the gRPC retry config once per RDS update. + pub retry_config: Option>, } /// Validated route match criteria. @@ -336,15 +367,9 @@ impl Resource for RouteConfigResource { } let mut routes = Vec::with_capacity(vh.routes.len()); - // gRFC A44: a `VirtualHost.retry_policy` applies to every route in the - // vhost that doesn't set its own; a route-level policy completely - // overrides it (values are not merged). Parse it once per vhost so all - // inheriting routes share the same `Arc`. - let vh_retry = vh.retry_policy.as_ref().map(|rp| { - Arc::new(GrpcRetrySharedConfig::from_route_retry( - &RouteRetryConfig::from_proto(rp), - )) - }); + // gRFC A44: routes inherit the virtual host's retry policy unless they + // set their own. Parse it once so inheriting routes share one `Arc`. + let vh_retry = vh.retry_policy.as_ref().map(parse_retry).transpose()?; for route in vh.routes { if let Some(validated_route) = validate_route(route, vh_retry.as_ref())? { routes.push(validated_route); @@ -369,12 +394,11 @@ impl Resource for RouteConfigResource { /// Returns `Ok(None)` for routes that should be silently skipped (query param matchers, /// unsupported cluster specifiers like `cluster_header`). /// -/// `vh_retry` is the enclosing virtual host's retry config (gRFC A44), if any. It is -/// used as the fallback for routes that don't set their own `RouteAction.retry_policy`; -/// a route-level policy takes precedence and completely overrides it. +/// `vh_retry` is the virtual host's retry config (gRFC A44), used as the fallback +/// when the route sets no `RouteAction.retry_policy` of its own. fn validate_route( route: envoy_types::pb::envoy::config::route::v3::Route, - vh_retry: Option<&Arc>, + vh_retry: Option<&Arc>, ) -> xds_client::Result> { let route_match = route .r#match @@ -392,23 +416,19 @@ fn validate_route( .ok_or_else(|| Error::Validation("route missing action field".into()))?; let validated_action; - let retry_config; + let route_retry; match action { - route::Action::Route(route_action) => { - // Parse and validate the route's retry policy before `route_action` - // is consumed. Deriving the gRPC config here — once, at RDS-validation - // time — is what keeps the request hot path free of parsing and - // allocation (see `RouteConfig::retry_config`). A route-level policy - // takes precedence over the virtual host's (gRFC A44); the vhost - // fallback is applied below. - retry_config = route_action.retry_policy.as_ref().map(|rp| { - let raw = RouteRetryConfig::from_proto(rp); - Arc::new(GrpcRetrySharedConfig::from_route_retry(&raw)) - }); + route::Action::Route(mut route_action) => { + // Take the retry policy before `route_action` is consumed, but parse + // it only once the route is known to be kept, so dropped routes cost + // no retry parsing (gRFC A44: a route-level policy overrides the + // virtual host's). + let retry_policy = route_action.retry_policy.take(); match validate_route_action(route_action)? { Some(action) => validated_action = action, None => return Ok(None), } + route_retry = retry_policy.as_ref().map(parse_retry).transpose()?; } // Per A28: action field must be "route", otherwise NACK. _ => { @@ -421,11 +441,8 @@ fn validate_route( Ok(Some(RouteConfig { match_criteria, action: validated_action, - // gRFC A44: fall back to the virtual host's retry policy when the route - // has none. `retry_config` is `Some` iff the route set its own policy, so - // `or_else` implements route-over-vhost precedence (a complete override, - // not a field merge). Inheriting routes share the vhost's `Arc`. - retry_config: retry_config.or_else(|| vh_retry.cloned()), + // gRFC A44: route-level policy wins; otherwise inherit the vhost's. + retry_config: route_retry.or_else(|| vh_retry.cloned()), })) } @@ -602,9 +619,10 @@ impl RouteConfigResource { mod tests { use super::*; use envoy_types::pb::envoy::config::route::v3::{ - RetryPolicy, RouteAction, VirtualHost, route::Action, route_action::ClusterSpecifier, + RetryPolicy, RouteAction, VirtualHost, retry_policy::RetryBackOff, route::Action, + route_action::ClusterSpecifier, }; - use envoy_types::pb::google::protobuf::UInt32Value; + use envoy_types::pb::google::protobuf::{Duration as ProtoDuration, UInt32Value}; fn make_route(prefix: &str, cluster: &str) -> envoy_types::pb::envoy::config::route::v3::Route { envoy_types::pb::envoy::config::route::v3::Route { @@ -647,6 +665,38 @@ mod tests { } } + fn proto_dur(seconds: i64, nanos: i32) -> ProtoDuration { + ProtoDuration { seconds, nanos } + } + + fn retry_policy_with_backoff( + base: Option, + max: Option, + ) -> RetryPolicy { + RetryPolicy { + retry_on: "unavailable".to_string(), + retry_back_off: Some(RetryBackOff { + base_interval: base, + max_interval: max, + }), + ..Default::default() + } + } + + fn validate_with_route_retry(retry: RetryPolicy) -> xds_client::Result { + let rc = RouteConfiguration { + name: "rc".to_string(), + virtual_hosts: vec![VirtualHost { + name: "vh1".to_string(), + domains: vec!["*".to_string()], + routes: vec![make_route_with_retry("/", "c1", retry)], + ..Default::default() + }], + ..Default::default() + }; + RouteConfigResource::validate(rc) + } + fn make_route_config(name: &str) -> RouteConfiguration { RouteConfiguration { name: name.to_string(), @@ -1160,4 +1210,72 @@ mod tests { let validated = RouteConfigResource::validate(rc).unwrap(); assert!(validated.virtual_hosts[0].routes[0].retry_config.is_none()); } + + // gRFC A44: the resource must be NACKed (validation error) when num_retries < 1 + // or a set base_interval/max_interval is not greater than zero. + + #[test] + fn test_retry_num_retries_zero_is_rejected() { + let err = validate_with_route_retry(retry_policy("unavailable", 0)).unwrap_err(); + assert!(matches!(err, Error::Validation(_))); + } + + #[test] + fn test_retry_base_interval_zero_is_rejected() { + let policy = retry_policy_with_backoff(Some(proto_dur(0, 0)), None); + assert!(validate_with_route_retry(policy).is_err()); + } + + #[test] + fn test_retry_base_interval_negative_is_rejected() { + let policy = retry_policy_with_backoff(Some(proto_dur(-1, 0)), None); + assert!(validate_with_route_retry(policy).is_err()); + } + + #[test] + fn test_retry_back_off_without_base_interval_is_rejected() { + // retry_back_off set but base_interval unset => base is 0 => rejected. + let policy = retry_policy_with_backoff(None, Some(proto_dur(1, 0))); + assert!(validate_with_route_retry(policy).is_err()); + } + + #[test] + fn test_retry_max_interval_zero_is_rejected() { + let policy = + retry_policy_with_backoff(Some(proto_dur(0, 100_000_000)), Some(proto_dur(0, 0))); + assert!(validate_with_route_retry(policy).is_err()); + } + + #[test] + fn test_retry_base_interval_below_1ms_is_accepted() { + // A44: values < 1ms are treated as 1ms (clamped), not rejected. + let policy = retry_policy_with_backoff(Some(proto_dur(0, 500_000)), None); + assert!(validate_with_route_retry(policy).is_ok()); + } + + #[test] + fn test_retry_valid_backoff_is_accepted() { + let mut policy = + retry_policy_with_backoff(Some(proto_dur(0, 100_000_000)), Some(proto_dur(1, 0))); + policy.num_retries = Some(UInt32Value { value: 2 }); + let validated = validate_with_route_retry(policy).expect("valid retry policy"); + assert!(validated.virtual_hosts[0].routes[0].retry_config.is_some()); + } + + #[test] + fn test_vhost_retry_invalid_is_rejected() { + // A virtual-host-level policy is validated the same way and NACKs on error. + let rc = RouteConfiguration { + name: "rc".to_string(), + virtual_hosts: vec![VirtualHost { + name: "vh1".to_string(), + domains: vec!["*".to_string()], + retry_policy: Some(retry_policy("unavailable", 0)), + routes: vec![make_route("/", "c1")], + ..Default::default() + }], + ..Default::default() + }; + assert!(RouteConfigResource::validate(rc).is_err()); + } } diff --git a/tonic-xds/src/xds/routing.rs b/tonic-xds/src/xds/routing.rs index be7a9b415..83915568d 100644 --- a/tonic-xds/src/xds/routing.rs +++ b/tonic-xds/src/xds/routing.rs @@ -43,7 +43,7 @@ use std::time::Duration; use arc_swap::ArcSwapOption; use tokio::sync::watch; -use crate::client::route::{AcquiredConfig, RouteDecision, RouteInput, Router}; +use crate::client::route::{AcquiredConfig, RouteDecision, RouteInput, Router, RoutingSnapshot}; use crate::common::async_util::AbortOnDrop; use crate::xds::cache::XdsCache; use crate::xds::resource::hash_policy::HashPolicyConfig; @@ -66,7 +66,7 @@ const DEFAULT_READY_TIMEOUT: Duration = Duration::from_secs(30); /// config is available, matching standard gRPC behavior where RPCs wait for the /// resolver's first update. Subsequent RPCs read the config lock-free. pub(crate) struct XdsRouter { - route_config: Arc>, + route_config: Arc>, ready_rx: watch::Receiver, _watch_task: AbortOnDrop, } @@ -85,7 +85,9 @@ impl XdsRouter { let handle = tokio::spawn(async move { let mut ready_tx = Some(ready_tx); while let Some(config) = watcher.next().await { - rc.store(Some(config)); + // Compile the retry configs once per RDS update, bundled with the + // resource so routing and retry read one consistent version. + rc.store(Some(Arc::new(RoutingSnapshot::new(config)))); // Signal readiness on the first config, then drop the sender. if let Some(tx) = ready_tx.take() { let _ = tx.send(true); @@ -100,7 +102,7 @@ impl XdsRouter { } /// The route config currently in effect, or `None` if none has arrived yet. - pub(crate) fn snapshot(&self) -> Option> { + pub(crate) fn snapshot(&self) -> Option> { self.route_config.load_full() } } @@ -126,7 +128,7 @@ impl Router for XdsRouter { fn route( &self, input: &RouteInput<'_>, - config: &RouteConfigResource, + config: &RoutingSnapshot, ) -> Result { resolve_route(config, input.authority, input.headers) } @@ -134,7 +136,7 @@ impl Router for XdsRouter { /// Resolve a route decision from the given config, authority, and headers. fn resolve_route( - rc: &RouteConfigResource, + rc: &RoutingSnapshot, authority: &str, headers: &http::HeaderMap, ) -> Result { @@ -150,13 +152,11 @@ fn resolve_route( .to_string(), }; - // Carry the matched route's shared retry config (if any) so the retry layer - // applies the config for exactly the route this request took (gRFC A44). This - // is a cheap `Arc` clone of an already-parsed, validated config; the retry - // layer instantiates a per-request policy from it with no allocation. Because - // routing and retry read the same config snapshot, both act on the same RDS - // version (no cross-layer skew). - let retry_config = route.retry_config.clone(); + // Look up the matched route's compiled retry config (if any) from the same + // snapshot, so the retry layer applies the config for exactly the route this + // request took (gRFC A44). This is a map lookup plus a cheap `Arc` clone; + // routing and retry read one RDS version, so there is no cross-layer skew. + let retry_config = rc.retry_for(route.retry_config.as_ref()); // gRFC A42 ring-hash request hash. The policy list is empty for now, so // `request_hash` resolves to `None` and the ring-hash picker falls back to a From 519b2806393b29929b12eb2637f12d0113d03f25 Mon Sep 17 00:00:00 2001 From: Yu Liu <60283975+LYZJU2019@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:44:09 -0700 Subject: [PATCH 3/4] docs(tonic-xds): trim redundant comments on the RDS retry path State the compile-once, single-Arc consistency, and hot-path rationale once on RoutingSnapshot, and keep the other retry-config sites terse. No code change. --- tonic-xds/src/client/channel.rs | 10 ++---- tonic-xds/src/client/retry.rs | 37 ++++++++-------------- tonic-xds/src/client/route.rs | 33 ++++++++----------- tonic-xds/src/xds/resource/route_config.rs | 32 +++++++------------ tonic-xds/src/xds/routing.rs | 8 ++--- 5 files changed, 42 insertions(+), 78 deletions(-) diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 5585bc24a..6a4011982 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -370,13 +370,9 @@ impl XdsChannelBuilder { ) -> XdsChannelGrpc { let router: Arc = Arc::new(XdsRouter::new(&cache)); - // Retry config is control-plane-driven from RDS, per route, compiled once - // per RDS update in the routing layer and carried on the request's - // `RouteDecision`; the retry layer reads that shared config `Arc` and - // instantiates a per-request policy from it, so the request hot path does - // no parsing or allocation. The default below is the fallback used when a - // request carries no route retry config (non-xDS callers, or a route with - // no retry policy). See [`RetryLayer`](crate::client::retry::RetryLayer). + // Fallback retry config for requests that carry no per-route config + // (non-xDS callers, or a route with no retry policy). Per-route configs + // come from RDS via the request's `RouteDecision`; see `RetryLayer`. let retry_layer = RetryLayer::new(Arc::new(GrpcRetrySharedConfig::default())); #[cfg(feature = "_tls-any")] diff --git a/tonic-xds/src/client/retry.rs b/tonic-xds/src/client/retry.rs index 74ade7a1d..fc153fd9f 100644 --- a/tonic-xds/src/client/retry.rs +++ b/tonic-xds/src/client/retry.rs @@ -314,8 +314,8 @@ impl RetryPolicy { Self::from_shared(Arc::new(RetrySharedConfig::new(config, classifier))) } - /// Instantiate per-request state from an already-shared config: an `Arc` - /// pointer clone plus a zero-field init. This is the hot path. + /// Instantiate per-request state from a shared config: a pointer clone plus + /// a zero-field init. pub(crate) fn from_shared(shared: Arc>) -> Self { Self { shared, @@ -324,8 +324,7 @@ impl RetryPolicy { } } - /// Consume the policy and return its shared config, discarding the - /// per-request state. + /// Consume the policy and return its shared config. pub(crate) fn into_shared(self) -> Arc> { self.shared } @@ -350,13 +349,11 @@ impl Default for RetryPolicy { } impl RetrySharedConfig { - /// Build a shared gRPC retry config from a route's [`RouteRetryConfig`] (RDS - /// `RouteAction.retry_policy`), parsing `retry_on` into [`tonic::Code`]s. - /// Unset Envoy fields fall back to [`RetryConfig`] defaults. - /// - /// Returns `None` when no `retry_on` condition maps to a gRPC status code - /// (gRFC A44): an empty set means "no retry policy" for the route, so it - /// falls back to the layer default rather than masking connection retries. + /// Build a shared gRPC config from a route's [`RouteRetryConfig`], mapping + /// `retry_on` to [`tonic::Code`]s; unset Envoy fields use [`RetryConfig`] + /// defaults. Returns `None` when no condition maps to a gRPC code, so the + /// route falls back to the layer default instead of masking connection + /// retries (gRFC A44). pub(crate) fn from_route_retry(retry: &RouteRetryConfig) -> Option { let retry_on = grpc_retry_on_codes(&retry.retry_on); if retry_on.is_empty() { @@ -443,16 +440,10 @@ pub(crate) type GrpcRetrySharedConfig = RetrySharedConfig; /// Tower [`Layer`] that wraps a gRPC service with retry support. /// -/// Converts the request body into a cloneable [`SharedBody`] and builds a fresh -/// [`tower::retry::Retry`] per request. The retry config is selected per request -/// from the route the request matched: the routing layer (just outside this one) -/// stamps the matched route's shared config into the request's [`RouteDecision`]. -/// Requests with no [`RouteDecision`] (non-xDS callers) or whose matched route +/// Builds a fresh [`tower::retry::Retry`] per request, selecting the config from +/// the matched route's [`RouteDecision`] (stamped by the routing layer just +/// outside). Requests with no [`RouteDecision`] (non-xDS callers) or whose route /// carries no retry policy use `fallback`. -/// -/// gRPC-specific because it reads the concrete [`RouteDecision`] extension; the -/// retry engine ([`RetryPolicy`], [`RetrySharedConfig`], [`RetryClassifier`]) -/// stays transport-agnostic. #[derive(Clone)] pub(crate) struct RetryLayer { /// Config used when a request carries no per-route retry config. @@ -460,8 +451,7 @@ pub(crate) struct RetryLayer { } impl RetryLayer { - /// Create a layer with the given `fallback` shared config, used when a - /// request carries no per-route retry config (see [`RetryLayer`]). + /// Create a layer with the given `fallback` config. pub(crate) fn new(fallback: Arc) -> Self { Self { fallback } } @@ -512,8 +502,7 @@ where } fn call(&mut self, request: Request) -> Self::Future { - // Use the shared config the routing layer stamped into the RouteDecision - // for the route this request matched, falling back when it carries none. + // Config the routing layer stamped for this request's route, or the fallback. let shared = request .extensions() .get::() diff --git a/tonic-xds/src/client/route.rs b/tonic-xds/src/client/route.rs index e9e95a922..27600d7eb 100644 --- a/tonic-xds/src/client/route.rs +++ b/tonic-xds/src/client/route.rs @@ -56,11 +56,9 @@ pub(crate) struct RouteDecision { #[allow(dead_code)] pub request_hash: Option, /// The matched route's compiled retry config (RDS `RouteAction.retry_policy`), - /// or `None` when the route specifies no retry. Resolved from the same - /// [`RoutingSnapshot`] the routing decision was made on, so the retry layer - /// applies the config for exactly the route this request took (gRFC A44). - /// Compiled once per RDS update and held behind an `Arc`, so the retry layer - /// instantiates a per-request policy from it with a pointer clone. + /// or `None` when the route sets no retry. Resolved from the [`RoutingSnapshot`] + /// this decision was made on, so the retry layer applies the config for exactly + /// the route the request took. pub retry_config: Option>, } @@ -76,29 +74,24 @@ pub trait PreRouteInterceptor: Send + Sync + 'static { fn on_request(&self, headers: &mut http::HeaderMap, metadata: &RouteConfigMetadata); } -/// A per-RDS-update routing snapshot: the validated [`RouteConfigResource`] -/// bundled with the gRPC retry configs compiled from it, once, when the snapshot -/// is installed. +/// The validated [`RouteConfigResource`] bundled with the gRPC retry configs +/// compiled from it, one per RDS update. /// -/// Holding both in one `Arc` lets the routing layer resolve a route and its -/// retry config from a single, consistent RDS version, and keeps the request hot -/// path to a map lookup plus an `Arc` clone (no parsing or allocation). -/// -/// Compiling the gRPC config here, rather than in the xDS resource layer, keeps -/// the resource types free of gRPC business logic. +/// Bundling both behind one `Arc` lets routing resolve a route and its retry +/// config from a single consistent RDS version, and keeps the request path to a +/// map lookup plus a pointer clone. Compiling here, rather than in the xDS +/// resource layer, keeps the resource types free of gRPC types. #[derive(Debug, Default)] pub(crate) struct RoutingSnapshot { resource: Arc, - /// Compiled retry config per route, keyed by the address of the route's - /// [`RouteRetryConfig`] `Arc`. Routes that share one config (vhost - /// inheritance) resolve to a single entry; routes whose `retry_on` maps to no - /// gRPC code have no entry (gRFC A44). + /// Compiled retry config, keyed by the identity (`Arc` address) of the + /// route's [`RouteRetryConfig`]. Routes that inherit a vhost policy share one + /// entry; a route whose `retry_on` maps to no gRPC code has none. retry: HashMap>, } impl RoutingSnapshot { - /// Builds a snapshot from a validated resource, compiling each distinct - /// route retry config once (gRFC A44). + /// Compiles each distinct route retry config once. pub(crate) fn new(resource: Arc) -> Self { let mut retry: HashMap> = HashMap::new(); for vhost in &resource.virtual_hosts { diff --git a/tonic-xds/src/xds/resource/route_config.rs b/tonic-xds/src/xds/resource/route_config.rs index f429c6343..ca2cd08c0 100644 --- a/tonic-xds/src/xds/resource/route_config.rs +++ b/tonic-xds/src/xds/resource/route_config.rs @@ -152,8 +152,7 @@ pub(crate) struct RouteConfigResource { } /// Validated Envoy retry settings (gRFC A44) parsed from a `RouteAction` or -/// `VirtualHost` `retry_policy` (RDS). A resource-layer type; the routing layer -/// compiles it into the gRPC retry config once per RDS update. +/// `VirtualHost` `retry_policy`. A transport-neutral resource-layer type. #[derive(Debug, Clone)] pub(crate) struct RouteRetryConfig { /// Envoy `retry_on` conditions, comma-separated (e.g. `"unavailable"`). @@ -223,10 +222,9 @@ impl RouteRetryConfig { /// Maximum `seconds` for a well-formed `google.protobuf.Duration`. const MAX_PROTO_DURATION_SECONDS: i64 = 315_576_000_000; -/// Convert a protobuf `Duration` to [`std::time::Duration`], returning `None` -/// for values outside the documented `google.protobuf.Duration` range -/// (negatives included). Out-of-range values are rejected here so an invalid -/// retry policy fails validation rather than overflowing later backoff math. +/// Convert a protobuf `Duration` to [`std::time::Duration`], returning `None` for +/// values outside the documented `google.protobuf.Duration` range. Rejected here +/// so an invalid retry policy fails validation rather than overflowing later. fn proto_duration(d: &envoy_types::pb::google::protobuf::Duration) -> Option { if !(0..=MAX_PROTO_DURATION_SECONDS).contains(&d.seconds) || !(0..=999_999_999).contains(&d.nanos) @@ -238,9 +236,7 @@ fn proto_duration(d: &envoy_types::pb::google::protobuf::Duration) -> Option xds_client::Result> { Ok(Arc::new(RouteRetryConfig::from_proto(rp)?)) } @@ -258,11 +254,9 @@ pub(crate) struct VirtualHostConfig { pub(crate) struct RouteConfig { pub match_criteria: RouteConfigMatch, pub action: RouteConfigAction, - /// Validated retry settings for this route (gRFC A44), or `None` when - /// neither the route nor its virtual host sets a policy. A route-level - /// policy completely overrides the virtual host's (values are not merged); - /// routes that inherit the vhost policy share one `Arc`. The routing layer - /// compiles this into the gRPC retry config once per RDS update. + /// Validated retry settings (gRFC A44): the route's own `RouteAction.retry_policy`, + /// else the inherited `VirtualHost.retry_policy` (route-level fully overrides, + /// no merge), else `None`. Routes that inherit share one `Arc`. pub retry_config: Option>, } @@ -394,8 +388,7 @@ impl Resource for RouteConfigResource { /// Returns `Ok(None)` for routes that should be silently skipped (query param matchers, /// unsupported cluster specifiers like `cluster_header`). /// -/// `vh_retry` is the virtual host's retry config (gRFC A44), used as the fallback -/// when the route sets no `RouteAction.retry_policy` of its own. +/// `vh_retry` is the virtual host's policy, inherited when the route sets none. fn validate_route( route: envoy_types::pb::envoy::config::route::v3::Route, vh_retry: Option<&Arc>, @@ -419,10 +412,8 @@ fn validate_route( let route_retry; match action { route::Action::Route(mut route_action) => { - // Take the retry policy before `route_action` is consumed, but parse - // it only once the route is known to be kept, so dropped routes cost - // no retry parsing (gRFC A44: a route-level policy overrides the - // virtual host's). + // Take the retry policy before `route_action` is consumed; parse it + // only if the route is kept, so dropped routes cost no parse. let retry_policy = route_action.retry_policy.take(); match validate_route_action(route_action)? { Some(action) => validated_action = action, @@ -441,7 +432,6 @@ fn validate_route( Ok(Some(RouteConfig { match_criteria, action: validated_action, - // gRFC A44: route-level policy wins; otherwise inherit the vhost's. retry_config: route_retry.or_else(|| vh_retry.cloned()), })) } diff --git a/tonic-xds/src/xds/routing.rs b/tonic-xds/src/xds/routing.rs index 83915568d..4a43e9d9b 100644 --- a/tonic-xds/src/xds/routing.rs +++ b/tonic-xds/src/xds/routing.rs @@ -85,8 +85,6 @@ impl XdsRouter { let handle = tokio::spawn(async move { let mut ready_tx = Some(ready_tx); while let Some(config) = watcher.next().await { - // Compile the retry configs once per RDS update, bundled with the - // resource so routing and retry read one consistent version. rc.store(Some(Arc::new(RoutingSnapshot::new(config)))); // Signal readiness on the first config, then drop the sender. if let Some(tx) = ready_tx.take() { @@ -152,10 +150,8 @@ fn resolve_route( .to_string(), }; - // Look up the matched route's compiled retry config (if any) from the same - // snapshot, so the retry layer applies the config for exactly the route this - // request took (gRFC A44). This is a map lookup plus a cheap `Arc` clone; - // routing and retry read one RDS version, so there is no cross-layer skew. + // Retry config for the matched route, from the same snapshot we routed on so + // retry and routing act on one RDS version. let retry_config = rc.retry_for(route.retry_config.as_ref()); // gRFC A42 ring-hash request hash. The policy list is empty for now, so From ca66d79195916ed4c52fcbad7aa2c0476c12220f Mon Sep 17 00:00:00 2001 From: Yu Liu <60283975+LYZJU2019@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:41:04 -0700 Subject: [PATCH 4/4] refactor(tonic-xds): take shared retry config in test channel helper Address review: build_grpc_channel_from_parts now takes Arc directly instead of a GrpcRetryPolicy it converts, and the now test-only retry-config imports live in the tests module. --- tonic-xds/src/client/channel.rs | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 6a4011982..786642608 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -46,8 +46,6 @@ use xds_client::{ ClientConfig, MetricsRecorder, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient, }; -#[cfg(test)] -use crate::client::retry::GrpcRetryPolicy; use crate::client::retry::{GrpcRetrySharedConfig, RetryLayer}; /// Configuration for building [`XdsChannel`] / [`XdsChannelGrpc`]. @@ -418,17 +416,17 @@ impl XdsChannelBuilder { } /// Builds an `XdsChannelGrpc` from the given router, cluster discovery, retry - /// policy, and optional pre-route interceptor. + /// config, and optional pre-route interceptor. #[cfg(test)] pub(crate) fn build_grpc_channel_from_parts( &self, router: Arc, discovery: Arc>>, - retry_policy: GrpcRetryPolicy, + retry: Arc, interceptor: Option>, ) -> XdsChannelGrpc { let routing_layer = XdsRoutingLayer::new(router, interceptor, self.authority()); - let retry_layer = RetryLayer::new(retry_policy.into_shared()); + let retry_layer = RetryLayer::new(retry); let cluster_registry = Arc::new(ClusterClientRegistryGrpc::new()); let lb_service = XdsLbService::new(cluster_registry, discovery); let inner = ServiceBuilder::new() @@ -464,7 +462,7 @@ mod tests { XdsChannelConfig::new(XdsUri::parse("xds:///test-service").unwrap()) } use crate::client::lb::{BoxDiscover, ClusterDiscovery}; - use crate::client::retry::GrpcRetryPolicy; + use crate::client::retry::GrpcRetrySharedConfig; use crate::client::route::RouteDecision; use crate::client::route::RouteInput; use crate::client::route::Router; @@ -616,7 +614,7 @@ mod tests { let xds_channel = xds_channel_builder.build_grpc_channel_from_parts( xds_manager.clone(), xds_manager.clone(), - GrpcRetryPolicy::default(), + Arc::new(GrpcRetrySharedConfig::default()), None, ); @@ -674,7 +672,7 @@ mod tests { #[tokio::test] async fn test_retry_once_on_unavailable() { - use crate::client::retry::{GrpcRetryClassifier, GrpcRetryPolicy, RetryConfig}; + use crate::client::retry::{GrpcRetryClassifier, RetryConfig}; use crate::testutil::grpc::spawn_fail_first_n_server; // Server fails the first request with UNAVAILABLE, succeeds on retry. @@ -685,15 +683,15 @@ mod tests { let servers = vec![server]; let xds_manager = Arc::new(MockXdsManager::from_test_servers(&servers)); - let retry_policy = GrpcRetryPolicy::new( + let retry = Arc::new(GrpcRetrySharedConfig::new( RetryConfig::new().num_retries(1), GrpcRetryClassifier::new(vec![tonic::Code::Unavailable]), - ); + )); let xds_channel = XdsChannelBuilder::new(test_config()).build_grpc_channel_from_parts( xds_manager.clone(), xds_manager.clone(), - retry_policy, + retry, None, ); @@ -796,7 +794,12 @@ mod tests { > = Arc::new(XdsClusterDiscovery::new(cache, GrpcMakeConnector::new())); let builder = XdsChannelBuilder::new(test_config()); - builder.build_grpc_channel_from_parts(router, discovery, GrpcRetryPolicy::default(), None) + builder.build_grpc_channel_from_parts( + router, + discovery, + Arc::new(GrpcRetrySharedConfig::default()), + None, + ) } /// Tests the full xDS stack (XdsRouter + XdsClusterDiscovery) with a