feat(tonic-xds): drive gRPC retry config from LDS control-plane updates

tonic-xds/src/client/channel.rs

@@ -26,11 +26,14 @@ use crate::client::cluster::ClusterClientRegistryGrpc;
 use crate::client::endpoint::{EndpointAddress, EndpointChannel};
 use crate::client::lb::{ClusterDiscovery, XdsLbService};
 use crate::client::route::{PreRouteInterceptor, Router, XdsRoutingLayer};
+use crate::common::async_util::AbortOnDrop;
 use crate::xds::bootstrap::{BootstrapConfig, BootstrapError};
-use crate::xds::cache::XdsCache;
+use crate::xds::cache::{CacheWatch, XdsCache};
 #[cfg(feature = "_tls-any")]
 use crate::xds::cert_provider::{CertProviderError, CertProviderRegistry, CertificateProvider};
 use crate::xds::cluster_discovery::{GrpcMakeConnector, XdsClusterDiscovery};
+use crate::xds::resource::ListenerResource;
+use crate::xds::resource::route_config::RouteConfigMetadata;
 use crate::xds::resource_manager::XdsResourceManager;
 use crate::xds::routing::XdsRouter;
 use crate::{TonicCallCredentials, XdsUri};
@@ -46,7 +49,13 @@ use xds_client::{
     ClientConfig, MetricsRecorder, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient,
 };
 
-use crate::client::retry::{GrpcRetryPolicy, RetryLayer};
+use crate::client::retry::{GrpcRetryPolicy, RetryConfig, RetryLayer, RetryPolicy};
+
+/// Extracts a [`RetryConfig`] from a Listener's [`RouteConfigMetadata`], returning
+/// `None` when no retry config is present. Registered via
+/// [`XdsChannelBuilder::with_retry_config_from_metadata`].
+type RetryConfigFromMetadata =
+    Arc<dyn Fn(&RouteConfigMetadata) -> Option<RetryConfig> + Send + Sync>;
 
 /// Configuration for building [`XdsChannel`] / [`XdsChannelGrpc`].
 #[derive(Clone, Debug)]
@@ -121,6 +130,10 @@ pub enum BuildError {
 struct XdsChannelResources {
     _resource_manager: XdsResourceManager,
     _xds_client: XdsClient,
+    /// Background task that hot-swaps the retry config from LDS updates. `None`
+    /// when no retry-config extractor is registered. Aborted when the last
+    /// channel clone drops.
+    _retry_watch_task: Option<AbortOnDrop>,
 }
 
 /// `XdsChannel` is an xDS-capable [`tower::Service`] implementation.
@@ -195,6 +208,7 @@ pub struct XdsChannelBuilder {
     config: Arc<XdsChannelConfig>,
     recorder: Option<Arc<dyn MetricsRecorder>>,
     pre_route: Option<Arc<dyn PreRouteInterceptor>>,
+    retry_config_from_metadata: Option<RetryConfigFromMetadata>,
     #[cfg(feature = "_tls-any")]
     cert_providers: HashMap<String, Arc<dyn CertificateProvider>>,
 }
@@ -216,6 +230,14 @@ impl Debug for XdsChannelBuilder {
                     .pre_route
                     .as_deref()
                     .map_or("None", |i| std::any::type_name_of_val(i)),
+            )
+            .field(
+                "retry_config_from_metadata",
+                &if self.retry_config_from_metadata.is_some() {
+                    "Some(..)"
+                } else {
+                    "None"
+                },
             );
         #[cfg(feature = "_tls-any")]
         s.field(
@@ -234,6 +256,7 @@ impl XdsChannelBuilder {
             config: Arc::new(config),
             recorder: None,
             pre_route: None,
+            retry_config_from_metadata: None,
             #[cfg(feature = "_tls-any")]
             cert_providers: HashMap::new(),
         }
@@ -266,6 +289,24 @@ impl XdsChannelBuilder {
         self
     }
 
+    /// Drives the retry configuration from the control plane.
+    ///
+    /// The provided closure extracts a [`RetryConfig`] from each Listener's
+    /// [`RouteConfigMetadata`](crate::RouteConfigMetadata) (LDS). A background task
+    /// applies it on every Listener update and hot-swaps the result into the retry
+    /// policy, so retry behavior tracks the control plane without rebuilding the
+    /// channel; returning `None` keeps the previous config.
+    ///
+    /// When unset, the channel uses a fixed default retry configuration.
+    #[must_use]
+    pub fn with_retry_config_from_metadata<F>(mut self, extract: F) -> Self
+    where
+        F: Fn(&RouteConfigMetadata) -> Option<RetryConfig> + Send + Sync + 'static,
+    {
+        self.retry_config_from_metadata = Some(Arc::new(extract));
+        self
+    }
+
     /// Registers a custom [`CertificateProvider`] under an xDS certificate
     /// provider instance name, resolved by CDS `UpstreamTlsContext` references.
     /// Shadows a bootstrap `file_watcher` instance of the same name.
@@ -367,6 +408,16 @@ impl XdsChannelBuilder {
         resource_manager: XdsResourceManager,
     ) -> XdsChannelGrpc {
         let router: Arc<dyn Router> = Arc::new(XdsRouter::new(&cache));
+
+        // Retry config is control-plane-driven: if the caller registered an
+        // extractor, seed the default policy and spawn a task that hot-swaps the
+        // config from LDS (Listener metadata) updates. `watch_listener()` must be
+        // taken before `cache` is moved into the cluster discovery.
+        let retry_policy = GrpcRetryPolicy::default();
+        let retry_watch_task = self.retry_config_from_metadata.clone().map(|extract| {
+            spawn_retry_config_watch(retry_policy.clone(), cache.watch_listener(), extract)
+        });
+
         #[cfg(feature = "_tls-any")]
         let discovery: Arc<
             dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
@@ -378,11 +429,11 @@ impl XdsChannelBuilder {
         let discovery: Arc<
             dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
         > = Arc::new(XdsClusterDiscovery::new(cache, GrpcMakeConnector::new()));
-        let retry_policy = GrpcRetryPolicy::default();
 
         let resources = Arc::new(XdsChannelResources {
             _resource_manager: resource_manager,
             _xds_client: xds_client,
+            _retry_watch_task: retry_watch_task,
         });
 
         let routing_layer = XdsRoutingLayer::new(router, self.pre_route.clone(), self.authority());
@@ -446,6 +497,32 @@ impl XdsChannelBuilder {
     }
 }
 
+/// Spawns a background task that keeps a [`RetryPolicy`]'s configuration in sync
+/// with the control plane. On each Listener (LDS) update, `extract` is applied to
+/// the listener's [`RouteConfigMetadata`]: a returned `Some(config)` is hot-swapped
+/// into the policy's shared config, while `None` leaves the previous config in
+/// place.
+///
+/// The returned [`AbortOnDrop`] handle stops the task when dropped, tying the
+/// task's lifetime to the channel resources that hold it.
+fn spawn_retry_config_watch<C>(
+    policy: RetryPolicy<C>,
+    mut watcher: CacheWatch<ListenerResource>,
+    extract: RetryConfigFromMetadata,
+) -> AbortOnDrop
+where
+    C: Send + 'static,
+{
+    let handle = tokio::spawn(async move {
+        while let Some(listener) = watcher.next().await {
+            if let Some(config) = extract(&listener.metadata) {
+                policy.update_config(config);
+            }
+        }
+    });
+    AbortOnDrop(handle)
+}
+
 #[cfg(test)]
 mod tests {
     use super::{XdsChannelBuilder, XdsChannelConfig};
@@ -921,4 +998,39 @@ mod tests {
         let _channel = builder.build_from_cache(cache, xds_client, resource_manager);
         // Construction should succeed without panicking.
     }
+
+    #[tokio::test]
+    async fn retry_config_hot_swaps_from_listener_metadata() {
+        use crate::client::retry::RetryConfig;
+        use crate::xds::resource::ListenerResource;
+        use crate::xds::resource::listener::RouteSource;
+        use crate::xds::resource::route_config::RouteConfigMetadata;
+
+        let cache = XdsCache::new();
+        let policy = GrpcRetryPolicy::default();
+        // Default config: a single retry.
+        assert_eq!(policy.load_config().num_retries, 1);
+
+        // The watch task recomputes the config from every Listener (LDS) update
+        // and hot-swaps it into the config shared across policy clones.
+        let extract: super::RetryConfigFromMetadata =
+            Arc::new(|_meta: &RouteConfigMetadata| Some(RetryConfig::new().num_retries(3)));
+        let _task =
+            super::spawn_retry_config_watch(policy.clone(), cache.watch_listener(), extract);
+
+        cache.update_listener(Arc::new(ListenerResource {
+            name: "listener".into(),
+            route_source: RouteSource::Rds("rc".into()),
+            metadata: RouteConfigMetadata::default(),
+        }));
+
+        // Yield until the spawned task observes the update and hot-swaps.
+        for _ in 0..100 {
+            if policy.load_config().num_retries == 3 {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+        assert_eq!(policy.load_config().num_retries, 3);
+    }
 }

tonic-xds/src/client/retry.rs

@@ -115,7 +115,7 @@ const MIN_BACKOFF: Duration = Duration::from_millis(1);
 /// - `max_interval` defaults to `10 * base_interval`.
 /// - `max_interval` must be >= `base_interval`; if not, it is clamped to `base_interval`.
 #[derive(Debug, Clone, PartialEq)]
-pub(crate) struct RetryBackoffConfig {
+pub struct RetryBackoffConfig {
     pub(crate) base_interval: Duration,
     pub(crate) max_interval: Duration,
     pub(crate) backoff_multiplier: f64,
@@ -125,7 +125,7 @@ impl RetryBackoffConfig {
     /// Create a new backoff config with the given `base_interval`.
     /// `max_interval` defaults to `10 * base_interval`.
     /// `backoff_multiplier` defaults to `2.0`.
-    pub(crate) fn new(base_interval: Duration) -> Self {
+    pub fn new(base_interval: Duration) -> Self {
         let base_interval = base_interval.max(MIN_BACKOFF);
         Self {
             max_interval: base_interval * 10,
@@ -136,14 +136,14 @@ impl RetryBackoffConfig {
 
     /// Set the maximum backoff interval.
     /// Values < 1ms are treated as 1ms. Values < `base_interval` are clamped to `base_interval`.
-    pub(crate) fn max_interval(mut self, max_interval: Duration) -> Self {
+    pub fn max_interval(mut self, max_interval: Duration) -> Self {
         let max_interval = max_interval.max(MIN_BACKOFF);
         self.max_interval = max_interval.max(self.base_interval);
         self
     }
 
     /// Set the backoff multiplier (default: 2.0).
-    pub(crate) fn backoff_multiplier(mut self, multiplier: f64) -> Self {
+    pub fn backoff_multiplier(mut self, multiplier: f64) -> Self {
         self.backoff_multiplier = multiplier;
         self
     }
@@ -155,7 +155,7 @@ impl Default for RetryBackoffConfig {
     }
 }
 
-/// Transport-agnostic retry knobs shared by every [`RetryClassifier`].
+/// Transport-agnostic retry knobs shared by every retry classifier.
 ///
 /// Built via [`RetryConfig::new`] with defaults, then customized via builder methods.
 ///
@@ -167,27 +167,27 @@ impl Default for RetryBackoffConfig {
 /// - `num_retries` must be >= 1. Values of 0 are clamped to 1.
 /// - `num_retries` is capped so total attempts (num_retries + 1) never exceed 5.
 #[derive(Debug, Clone)]
-pub(crate) struct RetryConfig {
+pub struct RetryConfig {
     pub(crate) num_retries: u32,
     pub(crate) retry_backoff: RetryBackoffConfig,
 }
 
 impl RetryConfig {
     /// Create a new retry config with defaults.
-    pub(crate) fn new() -> Self {
+    pub fn new() -> Self {
         Self::default()
     }
 
     /// Set the number of retries (total attempts = num_retries + 1).
     /// Values of 0 are clamped to 1. Values that would exceed 5 total attempts are capped.
-    pub(crate) fn num_retries(mut self, num_retries: u32) -> Self {
+    pub fn num_retries(mut self, num_retries: u32) -> Self {
         // Safety: clamp panics if min > max. Here min=1, max=MAX_ATTEMPTS-1=4 (const).
         self.num_retries = num_retries.clamp(1, MAX_ATTEMPTS - 1);
         self
     }
 
     /// Set the backoff configuration.
-    pub(crate) fn retry_backoff(mut self, backoff: RetryBackoffConfig) -> Self {
+    pub fn retry_backoff(mut self, backoff: RetryBackoffConfig) -> Self {
         self.retry_backoff = backoff;
         self
     }

tonic-xds/src/lib.rs

@@ -174,6 +174,7 @@ pub use client::channel::{
 pub use client::endpoint::{
     ClusterConfig, Connector, EndpointAddress, EndpointChannel, MakeConnector,
 };
+pub use client::retry::{RetryBackoffConfig, RetryConfig};
 pub use client::route::PreRouteInterceptor;
 pub use common::async_util::BoxFuture;
 pub use xds::bootstrap::{BootstrapConfig, BootstrapError};

tonic-xds/src/xds/cache.rs

@@ -33,7 +33,9 @@ use std::sync::Arc;
 use dashmap::DashMap;
 use tokio::sync::watch;
 
-use crate::xds::resource::{ClusterResource, EndpointsResource, RouteConfigResource};
+use crate::xds::resource::{
+    ClusterResource, EndpointsResource, ListenerResource, RouteConfigResource,
+};
 
 /// A wrapper around [`watch::Receiver`] that exposes only a single `next()`
 /// method, preventing misuse of the raw watch API.
@@ -123,6 +125,10 @@ impl<T> WatchMap<T> {
 /// - **Cluster layer**: calls [`watch_cluster`](Self::watch_cluster) to await
 ///   CDS readiness, then [`watch_endpoints`](Self::watch_endpoints) to track EDS updates.
 pub(crate) struct XdsCache {
+    /// Most recent Listener (LDS), surfaced so consumers can read listener-level
+    /// `metadata` (e.g. the control-plane-driven retry-config watcher).
+    listener_tx: watch::Sender<Option<Arc<ListenerResource>>>,
+
     /// Active route configuration (from LDS inline or RDS).
     route_config_tx: watch::Sender<Option<Arc<RouteConfigResource>>>,
 
@@ -136,14 +142,26 @@ pub(crate) struct XdsCache {
 impl XdsCache {
     /// Creates a new empty cache with no resources.
     pub(crate) fn new() -> Self {
+        let (listener_tx, _) = watch::channel(None);
         let (route_config_tx, _) = watch::channel(None);
         Self {
+            listener_tx,
             route_config_tx,
             clusters: WatchMap::new(),
             endpoints: WatchMap::new(),
         }
     }
 
+    /// Updates the active listener and notifies all watchers.
+    pub(crate) fn update_listener(&self, listener: Arc<ListenerResource>) {
+        self.listener_tx.send_replace(Some(listener));
+    }
+
+    /// Watches listener changes.
+    pub(crate) fn watch_listener(&self) -> CacheWatch<ListenerResource> {
+        CacheWatch::new(self.listener_tx.subscribe())
+    }
+
     /// Updates the active route configuration and notifies all watchers.
     pub(crate) fn update_route_config(&self, config: Arc<RouteConfigResource>) {
         self.route_config_tx.send_replace(Some(config));

tonic-xds/src/xds/resource/listener.rs

@@ -33,7 +33,7 @@ use prost::Message;
 use xds_client::resource::TypeUrl;
 use xds_client::{Error, Resource};
 
-use super::route_config::RouteConfigResource;
+use super::route_config::{RouteConfigMetadata, RouteConfigResource};
 
 /// How the listener obtains its route configuration.
 #[derive(Debug)]
@@ -47,11 +47,15 @@ pub(crate) enum RouteSource {
 /// Validated Listener resource.
 ///
 /// Extracts the route source from the
-/// `ApiListener` -> `HttpConnectionManager` -> `route_specifier` chain.
+/// `ApiListener` -> `HttpConnectionManager` -> `route_specifier` chain, and
+/// preserves the listener-level `metadata` (`filter_metadata`) so consumers can
+/// read config attached to the Listener (e.g. a retry-config extractor).
 #[derive(Debug)]
 pub(crate) struct ListenerResource {
     pub name: String,
     pub route_source: RouteSource,
+    /// Listener-level `metadata`. Empty when the Listener carried no metadata.
+    pub metadata: RouteConfigMetadata,
 }
 
 impl Resource for ListenerResource {
@@ -72,6 +76,14 @@ impl Resource for ListenerResource {
     fn validate(message: Self::Message) -> xds_client::Result<Self> {
         let name = message.name;
 
+        // Preserve listener-level metadata (filter_metadata / typed_filter_metadata)
+        // so downstream consumers (e.g. a retry-config extractor) can read config
+        // attached to the Listener.
+        let metadata = message
+            .metadata
+            .map(RouteConfigMetadata::from_proto)
+            .unwrap_or_default();
+
         // gRPC listeners must have an ApiListener.
         let api_listener = message
             .api_listener
@@ -98,13 +110,15 @@ impl Resource for ListenerResource {
                 Ok(ListenerResource {
                     name,
                     route_source: RouteSource::Rds(rds.route_config_name),
+                    metadata,
                 })
             }
             RouteSpecifier::RouteConfig(route_config) => {
                 let validated = RouteConfigResource::validate(route_config)?;
                 Ok(ListenerResource {
                     name,
                     route_source: RouteSource::Inline(validated),
+                    metadata,
                 })
             }
             RouteSpecifier::ScopedRoutes(_) => Err(Error::Validation(