tonic-xds/xds-client: client permanently stuck on stale endpoints after backend pod rotation — no keepalives or connect timeouts anywhere in the xDS path

tonic-xds/src/xds/cluster_discovery.rs

@@ -14,6 +14,7 @@
 //! connector is kept and a warning is logged.
 
 use std::sync::Arc;
+use std::time::Duration;
 
 use arc_swap::ArcSwap;
 use tokio::sync::mpsc;
@@ -38,6 +39,33 @@ use crate::xds::resource::security::ClusterSecurityConfig;
 /// Tower's LB layer.
 const DISCOVER_CHANNEL_CAPACITY: usize = 64;
 
+/// Timeout for establishing a connection to an endpoint.
+const ENDPOINT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// HTTP/2 keepalive PING interval for endpoint connections.
+const ENDPOINT_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(30);
+
+/// Time to wait for a keepalive PING ack before closing the connection.
+const ENDPOINT_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// Apply connection-liveness settings to a per-endpoint `Endpoint`.
+///
+/// Endpoint channels are created with `connect_lazy`, so a tonic `Channel`
+/// reports readiness independently of TCP connectivity. Without these
+/// settings a request routed to a black-holed address (e.g. a deleted pod IP,
+/// which Kubernetes drops without a RST) hangs on unanswered SYNs, and an
+/// established connection to a dead peer is never torn down — either way the
+/// endpoint stays "ready" to the LB and requests only die by the caller's
+/// deadline. The connect timeout and keepalives turn both cases into prompt
+/// transport errors instead.
+fn with_liveness_settings(endpoint: Endpoint) -> Endpoint {
+    endpoint
+        .connect_timeout(ENDPOINT_CONNECT_TIMEOUT)
+        .http2_keep_alive_interval(ENDPOINT_KEEP_ALIVE_INTERVAL)
+        .keep_alive_timeout(ENDPOINT_KEEP_ALIVE_TIMEOUT)
+        .keep_alive_while_idle(true)
+}
+
 /// xDS-backed cluster discovery.
 ///
 /// Resolves cluster names into endpoint change streams by watching the
@@ -189,9 +217,9 @@ impl Connector for PlaintextConnector {
         // EndpointAddress only holds validated Ipv4/Ipv6/Hostname + u16 port,
         // and its Display impl produces "ip:port" or "hostname:port". Prefixing
         // with "http://" always yields a valid URI, so from_shared cannot fail.
-        let channel = Endpoint::from_shared(format!("http://{addr}"))
-            .expect("EndpointAddress Display guarantees valid URI")
-            .connect_lazy();
+        let endpoint = Endpoint::from_shared(format!("http://{addr}"))
+            .expect("EndpointAddress Display guarantees valid URI");
+        let channel = with_liveness_settings(endpoint).connect_lazy();
         let svc = EndpointChannel::new(channel);
         Box::pin(async move { svc })
     }
@@ -284,8 +312,10 @@ impl Connector for TlsConnector {
         }
 
         let uri = format!("https://{addr}");
-        let endpoint = Endpoint::from_shared(uri.clone())
-            .expect("EndpointAddress Display guarantees valid URI");
+        let endpoint = with_liveness_settings(
+            Endpoint::from_shared(uri.clone())
+                .expect("EndpointAddress Display guarantees valid URI"),
+        );
 
         let channel = match endpoint.tls_config_with_verifier(tls_config, verifier) {
             Ok(ep) => ep.connect_lazy(),
@@ -298,9 +328,11 @@ impl Connector for TlsConnector {
                     error = %e, address = %addr,
                     "tls_config_with_verifier failed; non-TLS lazy fallback",
                 );
-                Endpoint::from_shared(uri)
-                    .expect("EndpointAddress Display guarantees valid URI")
-                    .connect_lazy()
+                with_liveness_settings(
+                    Endpoint::from_shared(uri)
+                        .expect("EndpointAddress Display guarantees valid URI"),
+                )
+                .connect_lazy()
             }
         };
         let svc = EndpointChannel::new(channel);

xds-client/src/transport/tonic.rs

@@ -10,6 +10,7 @@ use crate::transport::{Transport, TransportBuilder, TransportStream};
 use bytes::{Buf, BufMut, Bytes};
 use http::uri::PathAndQuery;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio_stream::StreamExt as _;
 use tonic::client::Grpc;
@@ -42,6 +43,23 @@ const ADS_PATH: &str =
 
 const ADS_CHANNEL_BUFFER_SIZE: usize = 16;
 
+/// Default timeout for establishing the TCP/TLS connection to the xDS server.
+const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// Default HTTP/2 keepalive PING interval on the ADS channel.
+///
+/// The ADS stream is mostly idle from the client's perspective (the server
+/// only pushes on resource changes), so without keepalives a half-open
+/// connection — e.g. after an xDS server restart where the RST/GOAWAY was
+/// lost, or a dropped conntrack/NAT entry — is undetectable: `recv()` on the
+/// stream pends forever and the client keeps serving its last known
+/// resources. Keepalives surface such connections as transport errors, which
+/// triggers the worker's reconnect + re-subscribe path.
+const DEFAULT_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(30);
+
+/// Default time to wait for a keepalive PING ack before closing the connection.
+const DEFAULT_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10);
+
 /// A codec that passes bytes through without serialization.
 ///
 /// This allows us to handle serialization in the xDS client layer
@@ -169,18 +187,38 @@ impl TonicTransport {
 /// let builder = TonicTransportBuilder::new()
 ///     .with_tls_config(ClientTlsConfig::new().with_enabled_roots());
 /// ```
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone)]
 pub struct TonicTransportBuilder {
     // Future extensions:
-    // - Connection timeout settings
-    // - Keep-alive configuration
     // - Connection pooling settings
     // - Per-server credential overrides (via ServerConfig.extensions)
     #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))]
     tls_config: Option<tonic::transport::ClientTlsConfig>,
 
     /// Per-stream call credentials for the ADS stream.
     call_creds: Option<Arc<dyn TonicCallCredentials>>,
+
+    /// Timeout for establishing the connection to the xDS server.
+    connect_timeout: Duration,
+
+    /// HTTP/2 keepalive PING interval; `None` disables keepalives.
+    keep_alive_interval: Option<Duration>,
+
+    /// Time to wait for a keepalive PING ack before closing the connection.
+    keep_alive_timeout: Duration,
+}
+
+impl Default for TonicTransportBuilder {
+    fn default() -> Self {
+        Self {
+            #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))]
+            tls_config: None,
+            call_creds: None,
+            connect_timeout: DEFAULT_CONNECT_TIMEOUT,
+            keep_alive_interval: Some(DEFAULT_KEEP_ALIVE_INTERVAL),
+            keep_alive_timeout: DEFAULT_KEEP_ALIVE_TIMEOUT,
+        }
+    }
 }
 
 impl TonicTransportBuilder {
@@ -189,6 +227,24 @@ impl TonicTransportBuilder {
         Self::default()
     }
 
+    /// Set the timeout for establishing the connection to the xDS server.
+    pub fn with_connect_timeout(mut self, timeout: Duration) -> Self {
+        self.connect_timeout = timeout;
+        self
+    }
+
+    /// Configure HTTP/2 keepalives on the ADS channel.
+    ///
+    /// A PING is sent every `interval` (even while the stream is idle); if no
+    /// ack arrives within `timeout` the connection is closed, surfacing a
+    /// transport error that triggers reconnect + re-subscription. Pass
+    /// `interval = None` to disable keepalives.
+    pub fn with_keep_alive(mut self, interval: Option<Duration>, timeout: Duration) -> Self {
+        self.keep_alive_interval = interval;
+        self.keep_alive_timeout = timeout;
+        self
+    }
+
     /// Set the TLS configuration for connections to the xDS server.
     ///
     /// When set, all connections created by this builder will use TLS
@@ -250,6 +306,14 @@ impl TransportBuilder for TonicTransportBuilder {
         let endpoint = Endpoint::from_shared(Self::ensure_secure_server_uri(server.uri(), secure))
             .map_err(|e| Error::Connection(e.to_string()))?;
 
+        let mut endpoint = endpoint.connect_timeout(self.connect_timeout);
+        if let Some(interval) = self.keep_alive_interval {
+            endpoint = endpoint
+                .http2_keep_alive_interval(interval)
+                .keep_alive_timeout(self.keep_alive_timeout)
+                .keep_alive_while_idle(true);
+        }
+
         #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))]
         let endpoint = match &self.tls_config {
             Some(tls) => endpoint