From 52bf9c8d2b7058dffec9776f2dbea6bbdb9e37b3 Mon Sep 17 00:00:00 2001 From: aryehlev Date: Wed, 29 Jul 2026 08:04:50 +0300 Subject: [PATCH 1/3] try fixing deadlock. --- tonic-xds/src/xds/resource_manager.rs | 158 +++++++++++ xds-client/src/client/watch.rs | 10 +- xds-client/src/client/worker.rs | 392 ++++++++++++++++++++++---- xds-client/src/transport/mod.rs | 93 ++++++ 4 files changed, 602 insertions(+), 51 deletions(-) diff --git a/tonic-xds/src/xds/resource_manager.rs b/tonic-xds/src/xds/resource_manager.rs index 7f646d0a0..2c8076512 100644 --- a/tonic-xds/src/xds/resource_manager.rs +++ b/tonic-xds/src/xds/resource_manager.rs @@ -617,4 +617,162 @@ mod tests { assert!(state.rds_watcher.is_some()); assert_eq!(state.rds_name.as_deref(), Some("rc")); } + + /// End-to-end replica of the production Istio hang: an RDS update that + /// references far more clusters than the xds-client worker's command + /// channel buffers (64), reconciled through the real cascade path — + /// `handle_rds` holds the event's `ProcessingDone` token while + /// `reconcile_clusters` awaits one `watch()` per cluster. + /// + /// Before xds-client gained ADS flow control, the worker sat inside + /// `handle_response` awaiting that token and stopped draining commands; + /// once the command channel filled, `reconcile_clusters` blocked on the + /// 65th watch and the client deadlocked. This test times out under that + /// behavior and completes under the fixed worker. + #[tokio::test] + async fn rds_referencing_many_clusters_reconciles_without_deadlock() { + use envoy_types::pb::envoy::config::route::v3::{ + Route, RouteAction, RouteConfiguration, RouteMatch, VirtualHost, route::Action, + route_action::ClusterSpecifier, route_match::PathSpecifier, + }; + use envoy_types::pb::envoy::service::discovery::v3::{ + DeltaDiscoveryRequest, DeltaDiscoveryResponse, DiscoveryRequest, DiscoveryResponse, + aggregated_discovery_service_server::{ + AggregatedDiscoveryService, AggregatedDiscoveryServiceServer, + }, + }; + use envoy_types::pb::google::protobuf::Any; + use prost::Message as _; + use std::pin::Pin; + use tokio_stream::{Stream, StreamExt as _}; + use xds_client::{ + ClientConfig, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, + XdsClient as RealXdsClient, + }; + + const RDS_TYPE_URL: &str = "type.googleapis.com/envoy.config.route.v3.RouteConfiguration"; + // Well past the worker's 64-slot command channel. + const CLUSTER_COUNT: usize = 100; + + /// ADS server that answers the first RDS subscription with one + /// RouteConfiguration referencing CLUSTER_COUNT clusters and ignores + /// everything else (ACKs, CDS subscriptions). + struct ManyClustersServer; + + #[tonic::async_trait] + impl AggregatedDiscoveryService for ManyClustersServer { + type StreamAggregatedResourcesStream = + Pin> + Send>>; + + async fn stream_aggregated_resources( + &self, + request: tonic::Request>, + ) -> Result, tonic::Status> + { + let mut inbound = request.into_inner(); + let outbound = async_stream::try_stream! { + let mut sent = false; + while let Some(Ok(req)) = inbound.next().await { + if req.type_url == RDS_TYPE_URL && !sent { + sent = true; + let rc = RouteConfiguration { + name: "rc-many".to_string(), + virtual_hosts: vec![VirtualHost { + name: "vh".to_string(), + domains: vec!["*".to_string()], + routes: (0..CLUSTER_COUNT) + .map(|i| Route { + r#match: Some(RouteMatch { + path_specifier: Some(PathSpecifier::Prefix( + format!("/svc-{i}"), + )), + ..Default::default() + }), + action: Some(Action::Route(RouteAction { + cluster_specifier: Some( + ClusterSpecifier::Cluster(format!( + "cluster-{i}" + )), + ), + ..Default::default() + })), + ..Default::default() + }) + .collect(), + ..Default::default() + }], + ..Default::default() + }; + yield DiscoveryResponse { + version_info: "1".to_string(), + type_url: RDS_TYPE_URL.to_string(), + nonce: "n1".to_string(), + resources: vec![Any { + type_url: RDS_TYPE_URL.to_string(), + value: rc.encode_to_vec(), + }], + ..Default::default() + }; + } + } + }; + Ok(tonic::Response::new(Box::pin(outbound))) + } + + type DeltaAggregatedResourcesStream = + Pin> + Send>>; + + async fn delta_aggregated_resources( + &self, + _request: tonic::Request>, + ) -> Result, tonic::Status> + { + Err(tonic::Status::unimplemented("delta not supported")) + } + } + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let incoming = async_stream::stream! { + loop { + match listener.accept().await { + Ok((socket, _)) => yield Ok::<_, std::io::Error>(socket), + Err(_) => break, + } + } + }; + tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(AggregatedDiscoveryServiceServer::new(ManyClustersServer)) + .serve_with_incoming(incoming) + .await + .unwrap(); + }); + + let config = ClientConfig::new(Node::new("test", "0"), format!("http://{addr}")); + let xds_client: RealXdsClient = RealXdsClient::builder( + config, + TonicTransportBuilder::new(), + ProstCodec, + TokioRuntime, + ) + .build(); + + let mut watcher = xds_client.watch::("rc-many").await; + let event = tokio::time::timeout(std::time::Duration::from_secs(10), watcher.next()) + .await + .expect("timed out waiting for the RDS update") + .expect("watcher closed"); + + let cache = test_cache(); + let mut state = CascadeState::new(); + tokio::time::timeout( + std::time::Duration::from_secs(10), + state.handle_rds(event, &xds_client, &cache), + ) + .await + .expect("deadlock: reconcile_clusters blocked while the worker awaited ProcessingDone"); + + assert_eq!(state.cds_watchers.len(), CLUSTER_COUNT); + } } diff --git a/xds-client/src/client/watch.rs b/xds-client/src/client/watch.rs index c2c1e3304..cb553dceb 100644 --- a/xds-client/src/client/watch.rs +++ b/xds-client/src/client/watch.rs @@ -36,9 +36,13 @@ use crate::resource::{DecodedResource, Resource}; /// A signal to indicate that processing of a resource event is complete. /// -/// The xDS client waits for this signal before sending ACK/NACK to the server. -/// This allows watchers to add cascading subscriptions (e.g. LDS -> RDS -> CDS -> EDS) -/// that will be included in the same ACK. +/// The xDS client does not read the next response from the ADS stream until +/// every watcher has signaled for the current one (ADS flow control, per +/// gRFC A88). This lets watchers apply an update — including adding cascading +/// subscriptions (e.g. LDS -> RDS -> CDS -> EDS) — before the next update can +/// arrive. It does *not* delay the ACK/NACK, and the worker keeps processing +/// watch/unwatch commands while waiting, so holding the token cannot deadlock +/// the client. /// /// # Automatic Signaling /// diff --git a/xds-client/src/client/worker.rs b/xds-client/src/client/worker.rs index 070a7cb20..eafa3b3a6 100644 --- a/xds-client/src/client/worker.rs +++ b/xds-client/src/client/worker.rs @@ -32,6 +32,8 @@ use std::collections::hash_map::Entry; use std::collections::{HashMap, HashSet}; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; @@ -573,6 +575,22 @@ pub(crate) struct AdsWorker { recorder: RecorderHandle, } +/// A watcher notification staged during response handling: the channel to +/// deliver on, the event (carrying its `ProcessingDone` token), and — for +/// events that participate in ADS flow control — the receiver that resolves +/// when the watcher finishes processing. +type Delivery = ( + mpsc::Sender>, + ResourceEvent, + Option>, +); + +/// In-flight watcher deliveries for one response: sends every staged event +/// (with backpressure) and resolves once all watchers signal `ProcessingDone`. +/// While unresolved it gates reading the *next* response, but nothing else — +/// see [`AdsWorker::run_connected`]. +type PendingDispatch = Pin + Send>>; + /// Outcome of a connected ADS session (see [`AdsWorker::run_connected`]). enum ConnectedOutcome { /// All `XdsClient` handles were dropped; the worker should shut down. @@ -762,14 +780,23 @@ where // A78 a stream that fails *after* receiving a response is not counted as // a server failure. let mut saw_response = false; + // Watcher deliveries for the last response, still in flight (ADS flow + // control, per gRFC A88). While `Some`, the next response is not read + // — but commands keep draining below. Awaiting the deliveries inline + // in `handle_response` instead would freeze the whole loop: a watcher + // that issues commands (e.g. cascading watches) while holding its + // `ProcessingDone` token could fill the command channel and deadlock + // against a worker that only resumes once that token drops. + let mut pending: Option = None; loop { tokio::select! { - result = stream.recv() => { + result = stream.recv(), if pending.is_none() => { match result { Ok(Some(bytes)) => { saw_response = true; - if self.handle_response(&mut stream, bytes).await.is_err() { - return ConnectedOutcome::Failed { saw_response }; + match self.handle_response(&mut stream, bytes).await { + Ok(deliveries) => pending = Self::dispatch_pending(deliveries), + Err(_) => return ConnectedOutcome::Failed { saw_response }, } } // Stream closed by server or errored; reconnect. @@ -777,6 +804,11 @@ where } } + // `unwrap` is safe: the branch is disabled when `pending` is `None`. + _ = async { pending.as_mut().unwrap().await }, if pending.is_some() => { + pending = None; + } + cmd = self.command_rx.recv() => { match cmd { Some(cmd) => { @@ -791,6 +823,29 @@ where } } + /// Build the in-flight delivery future for one response's staged watcher + /// notifications, or `None` when there is nothing to deliver. + fn dispatch_pending(deliveries: Vec) -> Option { + if deliveries.is_empty() { + return None; + } + Some(Box::pin(async move { + let mut done_rxs = Vec::with_capacity(deliveries.len()); + for (event_tx, event, done_rx) in deliveries { + // Backpressure: await if the watcher's channel is full. Send + // errors (watcher dropped) are ignored; the rejected event's + // `ProcessingDone` token resolves its receiver on drop. + let _ = event_tx.send(event).await; + if let Some(rx) = done_rx { + done_rxs.push(rx); + } + } + for rx in done_rxs { + let _ = rx.await; + } + })) + } + /// Handle a command, optionally sending network requests if connected. /// /// When `stream` is `None`, only state updates are performed (disconnected mode). @@ -974,18 +1029,24 @@ where /// - Valid resources are cached and dispatched to watchers /// - Invalid resources are cached as NACKed and errors sent to specific watchers /// - Missing resources (for types with ALL_RESOURCES_REQUIRED_IN_SOTW) are marked deleted + /// + /// Cache/state updates and the ACK/NACK happen here; the watcher + /// notifications are only *staged* and returned as [`Delivery`]s for + /// `run_connected` to drive, so a slow (or stuck) watcher delays reading + /// the next response — ADS flow control — without freezing command + /// processing. async fn handle_response( &mut self, stream: &mut S, bytes: Bytes, - ) -> Result<()> { + ) -> Result> { let response = self.codec.decode_response(bytes)?; let type_url = response.type_url.clone(); let (type_url_arc, decoder) = match self.type_states.get(&type_url) { Some(s) => (Arc::clone(&s.type_url), &s.decoder), None => { - return Ok(()); + return Ok(Vec::new()); } }; @@ -1028,27 +1089,25 @@ where .map(|r| r.name().to_string()) .collect(); - let mut processing_done_futures = self.dispatch_resources(&type_url, valid_resources).await; + // Stage watcher notifications instead of sending them here: the sends + // (and the ProcessingDone waits) happen in the returned deliveries, + // driven by `run_connected` concurrently with command processing. + // State/cache updates still happen synchronously below, so the ACK + // reflects the accepted config regardless of watcher progress. + let mut deliveries = Vec::new(); + + self.dispatch_resources(&mut deliveries, &type_url, valid_resources); // Only notify watchers for per-resource errors (where we know the name). // Top-level errors have no associated name, so no watcher to notify. for (resource_name, error) in &per_resource_errors { - self.notify_resource_error(&type_url, resource_name, error) - .await; + self.notify_resource_error(&mut deliveries, &type_url, resource_name, error); } // Detect deleted resources (per A53): // For resource types with ALL_RESOURCES_REQUIRED_IN_SOTW = true, // any previously-received resource not in this response is deleted. - let deleted_futures = self - .detect_deleted_resources(&type_url, &received_names) - .await; - processing_done_futures.extend(deleted_futures); - - // Wait for all watchers to finish processing. - for rx in processing_done_futures { - let _ = rx.await; - } + self.detect_deleted_resources(&mut deliveries, &type_url, &received_names); let has_errors = !top_level_errors.is_empty() || !per_resource_errors.is_empty(); if !has_errors { @@ -1057,7 +1116,7 @@ where if let Some(ts) = self.type_states.get_mut(&type_url) { ts.version_info = response.version_info.clone(); } - self.send_ack(stream, &response).await + self.send_ack(stream, &response).await?; } else { // Build NACK message combining both error categories let mut error_parts = Vec::new(); @@ -1076,21 +1135,23 @@ where } self.send_nack(stream, &response, error_parts.join("; ")) - .await + .await?; } + + Ok(deliveries) } - /// Dispatch decoded resources to watchers and update cache. + /// Update the cache from decoded resources and stage watcher deliveries. /// - /// Returns futures that resolve when watchers signal ProcessingDone. - /// Uses backpressure: waits if a watcher's channel is full. - async fn dispatch_resources( + /// The staged events carry `ProcessingDone` receivers that gate reading + /// the next response (ADS flow control); the sends themselves happen in + /// the [`PendingDispatch`] future, with backpressure on full channels. + fn dispatch_resources( &mut self, + deliveries: &mut Vec, type_url: &str, resources: Vec, - ) -> Vec> { - let mut processing_done_futures = Vec::new(); - + ) { let watcher_info: Vec<_> = match self.type_states.get_mut(type_url) { Some(s) => { for resource in &resources { @@ -1107,7 +1168,7 @@ where .map(|(id, entry)| (*id, entry.event_tx.clone(), entry.subscription.clone())) .collect() } - None => return processing_done_futures, + None => return, }; // Cancel resource timers for received resources (gRFC A57). @@ -1127,22 +1188,24 @@ where result: Ok(Arc::clone(&resource)), done, }; - // Use backpressure: await if channel is full. - // Ignore send errors (watcher dropped). - let _ = event_tx.send(event).await; - processing_done_futures.push(rx); + deliveries.push((event_tx, event, Some(rx))); } } } - - processing_done_futures } - /// Notify watchers of a validation error for a specific resource. + /// Stage validation-error notifications for a specific resource. /// /// Per gRFC A46/A88, errors are routed only to watchers interested in - /// that specific resource (plus wildcard watchers). - async fn notify_resource_error(&mut self, type_url: &str, resource_name: &str, error: &str) { + /// that specific resource (plus wildcard watchers). Error events do not + /// gate flow control (no `ProcessingDone` receiver is staged). + fn notify_resource_error( + &mut self, + deliveries: &mut Vec, + type_url: &str, + resource_name: &str, + error: &str, + ) { let type_state = match self.type_states.get_mut(type_url) { Some(s) => s, None => return, @@ -1166,29 +1229,29 @@ where result: Err(Error::Validation(error.to_string())), done, }; - let _ = event_tx.send(event).await; + deliveries.push((event_tx, event, None)); } } - /// Detect resources that were deleted (present in cache but not in response). + /// Detect resources that were deleted (present in cache but not in response) + /// and stage the deletion notifications. /// /// Per gRFC A53, for resource types with ALL_RESOURCES_REQUIRED_IN_SOTW = true, /// if a previously-received resource is absent from a new SotW response, /// it is treated as deleted. - async fn detect_deleted_resources( + fn detect_deleted_resources( &mut self, + deliveries: &mut Vec, type_url: &str, received_names: &HashSet, - ) -> Vec> { - let mut processing_done_futures = Vec::new(); - + ) { let type_state = match self.type_states.get_mut(type_url) { Some(s) => s, - None => return processing_done_futures, + None => return, }; if !type_state.all_resources_required_in_sotw { - return processing_done_futures; + return; } let deleted_names: Vec = type_state @@ -1211,8 +1274,7 @@ where result: Err(Error::ResourceDoesNotExist), done, }; - let _ = event_tx.send(event).await; - processing_done_futures.push(rx); + deliveries.push((event_tx, event, Some(rx))); } } @@ -1220,8 +1282,6 @@ where let counts = type_state.resource_state_counts(); self.recorder .sync_resource_counts(&type_state.type_url, &counts); - - processing_done_futures } /// Send an ACK for a response. @@ -1551,3 +1611,239 @@ mod tests { assert_eq!(gauge_for(&events, "acked"), Some(0)); } } + +/// Regression tests for the worker's ADS flow control: watcher deliveries and +/// `ProcessingDone` waits must not freeze the event loop (deadlock with +/// watchers that issue commands while holding the token) and must gate +/// reading the next response. +#[cfg(test)] +mod flow_control_tests { + use std::time::Duration; + + use bytes::Bytes; + + use crate::client::config::ClientConfig; + use crate::client::watch::ResourceEvent; + use crate::codec::XdsCodec; + use crate::error::Result; + use crate::message::{DiscoveryRequest, DiscoveryResponse, Node, ResourceAny}; + use crate::resource::{Resource, TypeUrl}; + use crate::runtime::tokio::TokioRuntime; + use crate::transport::mock::{MockServer, mock_transport}; + use crate::{XdsClient, error::Error}; + + const TEST_TYPE_URL: &str = "type.googleapis.com/test.Resource"; + + /// Minimal resource: the message is the resource name itself. + #[derive(Debug, Clone)] + struct TestResource; + + impl Resource for TestResource { + type Message = String; + const TYPE_URL: TypeUrl = TypeUrl::new(TEST_TYPE_URL); + const ALL_RESOURCES_REQUIRED_IN_SOTW: bool = false; + + fn deserialize(bytes: Bytes) -> Result { + String::from_utf8(bytes.to_vec()).map_err(|e| Error::Validation(e.to_string())) + } + + fn name(message: &Self::Message) -> &str { + message + } + + fn validate(_message: Self::Message) -> Result { + Ok(Self) + } + } + + /// Line-based codec: `type_url \n version \n nonce \n name,name,...`. + struct FakeCodec; + + impl XdsCodec for FakeCodec { + fn encode_request(&self, request: &DiscoveryRequest<'_>) -> Result { + Ok(Bytes::from(format!( + "{}\n{}\n{}\n{}", + request.type_url, + request.version_info, + request.response_nonce, + request.resource_names.join(",") + ))) + } + + fn decode_response(&self, bytes: Bytes) -> Result { + let text = + String::from_utf8(bytes.to_vec()).map_err(|e| Error::Validation(e.to_string()))?; + let mut lines = text.split('\n'); + let type_url = lines.next().unwrap_or_default().to_string(); + let version_info = lines.next().unwrap_or_default().to_string(); + let nonce = lines.next().unwrap_or_default().to_string(); + let resources = lines + .next() + .unwrap_or_default() + .split(',') + .filter(|n| !n.is_empty()) + .map(|name| ResourceAny { + type_url: type_url.clone(), + value: Bytes::from(name.to_string()), + }) + .collect(); + Ok(DiscoveryResponse { + version_info, + resources, + type_url, + nonce, + }) + } + } + + fn response(version: &str, nonce: &str, names: &[&str]) -> Bytes { + Bytes::from(format!( + "{TEST_TYPE_URL}\n{version}\n{nonce}\n{}", + names.join(",") + )) + } + + /// (version_info, response_nonce) of an encoded request. + fn parse_request(bytes: &Bytes) -> (String, String) { + let text = String::from_utf8(bytes.to_vec()).unwrap(); + let mut lines = text.split('\n'); + let _type_url = lines.next().unwrap_or_default(); + let version = lines.next().unwrap_or_default().to_string(); + let nonce = lines.next().unwrap_or_default().to_string(); + (version, nonce) + } + + /// Client watching `res-0` with an established mock stream, its initial + /// request already drained. + async fn connected_client() -> ( + XdsClient, + crate::client::watch::ResourceWatcher, + MockServer, + ) { + let (builder, mut servers) = mock_transport(); + let config = ClientConfig::new(Node::new("test", "0"), "mock:///xds"); + let client = XdsClient::builder(config, builder, FakeCodec, TokioRuntime).build(); + + let watcher = client.watch::("res-0").await; + let mut server = tokio::time::timeout(Duration::from_secs(5), servers.recv()) + .await + .expect("timed out waiting for stream") + .expect("transport dropped"); + let _initial = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("timed out waiting for initial request") + .expect("stream closed"); + (client, watcher, server) + } + + /// A watcher that issues more commands than the command channel buffers + /// while holding its `ProcessingDone` token must not deadlock the worker. + /// + /// Before the flow-control fix the worker awaited the token inside + /// `handle_response`, so it stopped draining commands; once the channel + /// filled, watcher and worker waited on each other forever. + #[tokio::test] + async fn commands_drain_while_processing_done_is_held() { + let (client, mut watcher, server) = connected_client().await; + + server + .responses + .send(Ok(Some(response("1", "n1", &["res-0"])))) + .unwrap(); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for event") + .expect("watcher closed"); + let ResourceEvent::ResourceChanged { + result: Ok(_), + done, + } = event + else { + panic!("expected ResourceChanged(Ok)"); + }; + + // Well past COMMAND_CHANNEL_BUFFER_SIZE (64). + let mut extra = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), async { + for i in 1..=100 { + extra.push(client.watch::(format!("res-{i}")).await); + } + }) + .await + .expect("deadlock: commands not drained while ProcessingDone was held"); + + // The worker is still healthy end-to-end: it reads the next response + // and delivers it. + drop(done); + server + .responses + .send(Ok(Some(response("2", "n2", &["res-0"])))) + .unwrap(); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for second event") + .expect("watcher closed"); + assert!(matches!( + event, + ResourceEvent::ResourceChanged { result: Ok(_), .. } + )); + } + + /// The ACK goes out as soon as the response is validated and cached, and + /// the *next* response is not delivered until the previous one's + /// `ProcessingDone` tokens drop (ADS flow control). + #[tokio::test] + async fn next_response_gated_until_processing_done() { + let (_client, mut watcher, mut server) = connected_client().await; + + server + .responses + .send(Ok(Some(response("1", "n1", &["res-0"])))) + .unwrap(); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for event") + .expect("watcher closed"); + let ResourceEvent::ResourceChanged { + result: Ok(_), + done, + } = event + else { + panic!("expected ResourceChanged(Ok)"); + }; + + // ACK is not gated on ProcessingDone. + let ack = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("ACK not sent while ProcessingDone was held") + .expect("stream closed"); + assert_eq!(parse_request(&ack), ("1".to_string(), "n1".to_string())); + + // The next response is gated on ProcessingDone. + server + .responses + .send(Ok(Some(response("2", "n2", &["res-0"])))) + .unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(200), watcher.next()) + .await + .is_err(), + "second response delivered while the first was still being processed" + ); + + drop(done); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for second event") + .expect("watcher closed"); + assert!(matches!( + event, + ResourceEvent::ResourceChanged { result: Ok(_), .. } + )); + let ack = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("second ACK not sent") + .expect("stream closed"); + assert_eq!(parse_request(&ack), ("2".to_string(), "n2".to_string())); + } +} diff --git a/xds-client/src/transport/mod.rs b/xds-client/src/transport/mod.rs index 033ff8558..2f509c695 100644 --- a/xds-client/src/transport/mod.rs +++ b/xds-client/src/transport/mod.rs @@ -122,3 +122,96 @@ pub trait TransportBuilder: Send + Sync + 'static { // - `fn close(&self, server: &ServerConfig)` for explicit connection cleanup // - Metrics/observability hooks } + +/// In-crate mock transport for worker tests. +/// +/// Lives here because [`TransportStream`] is sealed: test code outside this +/// module cannot implement it. +#[cfg(test)] +pub(crate) mod mock { + use super::{Transport, TransportBuilder, TransportStream, sealed}; + use crate::client::config::ServerConfig; + use crate::error::{Error, Result}; + use bytes::Bytes; + use tokio::sync::mpsc; + + /// Test-side handle to one mock ADS stream. + pub(crate) struct MockServer { + /// Requests the worker sent (initial requests, ACKs, subscription changes). + pub(crate) requests: mpsc::UnboundedReceiver, + /// Responses to push to the worker. + pub(crate) responses: mpsc::UnboundedSender>>, + } + + /// Returns a transport builder for the worker plus the receiver on which + /// the test obtains a [`MockServer`] for every stream the worker opens. + pub(crate) fn mock_transport() -> (MockTransportBuilder, mpsc::UnboundedReceiver) { + let (servers_tx, servers_rx) = mpsc::unbounded_channel(); + ( + MockTransportBuilder { + servers: servers_tx, + }, + servers_rx, + ) + } + + pub(crate) struct MockTransportBuilder { + servers: mpsc::UnboundedSender, + } + + impl TransportBuilder for MockTransportBuilder { + type Transport = MockTransport; + + async fn build(&self, _server: &ServerConfig) -> Result { + Ok(MockTransport { + servers: self.servers.clone(), + }) + } + } + + pub(crate) struct MockTransport { + servers: mpsc::UnboundedSender, + } + + impl Transport for MockTransport { + type Stream = MockStream; + + async fn new_stream(&self, initial_requests: Vec) -> Result { + let (req_tx, req_rx) = mpsc::unbounded_channel(); + let (resp_tx, resp_rx) = mpsc::unbounded_channel(); + for request in initial_requests { + let _ = req_tx.send(request); + } + let _ = self.servers.send(MockServer { + requests: req_rx, + responses: resp_tx, + }); + Ok(MockStream { + requests: req_tx, + responses: resp_rx, + }) + } + } + + pub(crate) struct MockStream { + requests: mpsc::UnboundedSender, + responses: mpsc::UnboundedReceiver>>, + } + + impl sealed::Sealed for MockStream {} + + impl TransportStream for MockStream { + async fn send(&mut self, request: Bytes) -> Result<()> { + self.requests + .send(request) + .map_err(|_| Error::Connection("mock stream closed".into())) + } + + async fn recv(&mut self) -> Result> { + match self.responses.recv().await { + Some(result) => result, + None => Ok(None), + } + } + } +} From 56d16928d1f09754dc3cc481b90b9b0d01f5d0f2 Mon Sep 17 00:00:00 2001 From: aryehlev Date: Thu, 30 Jul 2026 10:52:05 +0300 Subject: [PATCH 2/3] use XdsTestControlPlaneService. --- tonic-xds/src/xds/resource_manager.rs | 135 ++++---------------------- xds-test-util/src/config.rs | 97 ++++++++++++++++++ 2 files changed, 118 insertions(+), 114 deletions(-) diff --git a/tonic-xds/src/xds/resource_manager.rs b/tonic-xds/src/xds/resource_manager.rs index 2c8076512..98543cc10 100644 --- a/tonic-xds/src/xds/resource_manager.rs +++ b/tonic-xds/src/xds/resource_manager.rs @@ -631,127 +631,34 @@ mod tests { /// behavior and completes under the fixed worker. #[tokio::test] async fn rds_referencing_many_clusters_reconciles_without_deadlock() { - use envoy_types::pb::envoy::config::route::v3::{ - Route, RouteAction, RouteConfiguration, RouteMatch, VirtualHost, route::Action, - route_action::ClusterSpecifier, route_match::PathSpecifier, - }; - use envoy_types::pb::envoy::service::discovery::v3::{ - DeltaDiscoveryRequest, DeltaDiscoveryResponse, DiscoveryRequest, DiscoveryResponse, - aggregated_discovery_service_server::{ - AggregatedDiscoveryService, AggregatedDiscoveryServiceServer, - }, - }; - use envoy_types::pb::google::protobuf::Any; - use prost::Message as _; - use std::pin::Pin; - use tokio_stream::{Stream, StreamExt as _}; + use std::time::Duration; use xds_client::{ ClientConfig, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient as RealXdsClient, }; + use xds_test_util::{XdsTestControlPlaneService, config}; - const RDS_TYPE_URL: &str = "type.googleapis.com/envoy.config.route.v3.RouteConfiguration"; // Well past the worker's 64-slot command channel. const CLUSTER_COUNT: usize = 100; - /// ADS server that answers the first RDS subscription with one - /// RouteConfiguration referencing CLUSTER_COUNT clusters and ignores - /// everything else (ACKs, CDS subscriptions). - struct ManyClustersServer; - - #[tonic::async_trait] - impl AggregatedDiscoveryService for ManyClustersServer { - type StreamAggregatedResourcesStream = - Pin> + Send>>; - - async fn stream_aggregated_resources( - &self, - request: tonic::Request>, - ) -> Result, tonic::Status> - { - let mut inbound = request.into_inner(); - let outbound = async_stream::try_stream! { - let mut sent = false; - while let Some(Ok(req)) = inbound.next().await { - if req.type_url == RDS_TYPE_URL && !sent { - sent = true; - let rc = RouteConfiguration { - name: "rc-many".to_string(), - virtual_hosts: vec![VirtualHost { - name: "vh".to_string(), - domains: vec!["*".to_string()], - routes: (0..CLUSTER_COUNT) - .map(|i| Route { - r#match: Some(RouteMatch { - path_specifier: Some(PathSpecifier::Prefix( - format!("/svc-{i}"), - )), - ..Default::default() - }), - action: Some(Action::Route(RouteAction { - cluster_specifier: Some( - ClusterSpecifier::Cluster(format!( - "cluster-{i}" - )), - ), - ..Default::default() - })), - ..Default::default() - }) - .collect(), - ..Default::default() - }], - ..Default::default() - }; - yield DiscoveryResponse { - version_info: "1".to_string(), - type_url: RDS_TYPE_URL.to_string(), - nonce: "n1".to_string(), - resources: vec![Any { - type_url: RDS_TYPE_URL.to_string(), - value: rc.encode_to_vec(), - }], - ..Default::default() - }; - } - } - }; - Ok(tonic::Response::new(Box::pin(outbound))) - } - - type DeltaAggregatedResourcesStream = - Pin> + Send>>; - - async fn delta_aggregated_resources( - &self, - _request: tonic::Request>, - ) -> Result, tonic::Status> - { - Err(tonic::Status::unimplemented("delta not supported")) - } - } - - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - let incoming = async_stream::stream! { - loop { - match listener.accept().await { - Ok((socket, _)) => yield Ok::<_, std::io::Error>(socket), - Err(_) => break, - } - } - }; - tokio::spawn(async move { - tonic::transport::Server::builder() - .add_service(AggregatedDiscoveryServiceServer::new(ManyClustersServer)) - .serve_with_incoming(incoming) - .await - .unwrap(); - }); - - let config = ClientConfig::new(Node::new("test", "0"), format!("http://{addr}")); + let control_plane = XdsTestControlPlaneService::new() + .start() + .await + .expect("control plane failed to start"); + control_plane.get_service().set_xds_config( + &config::AdsTypeUrl::Rds, + HashMap::from([( + "rc-many".to_string(), + config::build_route_config_with_cluster_count("rc-many", CLUSTER_COUNT), + )]), + ); + + let client_config = ClientConfig::new( + Node::new("test", "0"), + format!("http://{}", control_plane.addr()), + ); let xds_client: RealXdsClient = RealXdsClient::builder( - config, + client_config, TonicTransportBuilder::new(), ProstCodec, TokioRuntime, @@ -759,7 +666,7 @@ mod tests { .build(); let mut watcher = xds_client.watch::("rc-many").await; - let event = tokio::time::timeout(std::time::Duration::from_secs(10), watcher.next()) + let event = tokio::time::timeout(Duration::from_secs(10), watcher.next()) .await .expect("timed out waiting for the RDS update") .expect("watcher closed"); @@ -767,7 +674,7 @@ mod tests { let cache = test_cache(); let mut state = CascadeState::new(); tokio::time::timeout( - std::time::Duration::from_secs(10), + Duration::from_secs(10), state.handle_rds(event, &xds_client, &cache), ) .await diff --git a/xds-test-util/src/config.rs b/xds-test-util/src/config.rs index 45207df04..f211917ee 100644 --- a/xds-test-util/src/config.rs +++ b/xds-test-util/src/config.rs @@ -127,6 +127,48 @@ pub fn build_route_config(name: &str, cluster: &str) -> RouteConfiguration { } } +/// Builds a route configuration referencing `count` generated clusters +/// (`cluster-0` .. `cluster-{count-1}`), one prefix route per cluster. +/// +/// With `count = 1` this is the generated-name equivalent of +/// [`build_route_config`]. Larger counts mirror control planes that share one +/// route config across many services (e.g. Istio's per-port route configs). +pub fn build_route_config_with_cluster_count(name: &str, count: usize) -> RouteConfiguration { + let clusters: Vec = (0..count).map(|i| format!("cluster-{i}")).collect(); + build_route_config_with_clusters(name, &clusters) +} + +/// Builds a route configuration with one prefix route per cluster +/// (`/` -> that cluster), all in one virtual host. +/// +/// Mirrors control planes that share one route config across many services +/// (e.g. Istio's per-port route configs). +pub fn build_route_config_with_clusters(name: &str, clusters: &[String]) -> RouteConfiguration { + RouteConfiguration { + name: name.to_string(), + virtual_hosts: vec![VirtualHost { + name: "default".to_string(), + domains: vec!["*".to_string()], + routes: clusters + .iter() + .map(|cluster| Route { + r#match: Some(RouteMatch { + path_specifier: Some(PathSpecifier::Prefix(format!("/{cluster}"))), + ..Default::default() + }), + action: Some(Action::Route(RouteAction { + cluster_specifier: Some(ClusterSpecifier::Cluster(cluster.to_string())), + ..Default::default() + })), + ..Default::default() + }) + .collect(), + ..Default::default() + }], + ..Default::default() + } +} + /// Builds an EDS-discovered cluster named `name`. pub fn build_cluster(name: &str) -> Cluster { Cluster { @@ -172,3 +214,58 @@ fn lb_endpoint(host: &str, port: u16) -> LbEndpoint { ..Default::default() } } + +#[cfg(test)] +mod tests { + use super::*; + + fn route_clusters(rc: &RouteConfiguration) -> Vec<(String, String)> { + rc.virtual_hosts[0] + .routes + .iter() + .map(|r| { + let prefix = match r.r#match.as_ref().and_then(|m| m.path_specifier.as_ref()) { + Some(PathSpecifier::Prefix(p)) => p.clone(), + other => panic!("expected prefix match, got {other:?}"), + }; + let cluster = match r.action.as_ref() { + Some(Action::Route(a)) => match a.cluster_specifier.as_ref() { + Some(ClusterSpecifier::Cluster(c)) => c.clone(), + other => panic!("expected cluster specifier, got {other:?}"), + }, + other => panic!("expected route action, got {other:?}"), + }; + (prefix, cluster) + }) + .collect() + } + + #[test] + fn route_config_with_clusters_builds_one_route_per_cluster() { + let clusters = vec!["a".to_string(), "b".to_string()]; + let rc = build_route_config_with_clusters("rc", &clusters); + assert_eq!(rc.name, "rc"); + assert_eq!( + route_clusters(&rc), + vec![ + ("/a".to_string(), "a".to_string()), + ("/b".to_string(), "b".to_string()), + ] + ); + } + + #[test] + fn route_config_with_cluster_count_generates_names() { + let rc = build_route_config_with_cluster_count("rc", 3); + let routes = route_clusters(&rc); + assert_eq!(routes.len(), 3); + assert_eq!( + routes[0], + ("/cluster-0".to_string(), "cluster-0".to_string()) + ); + assert_eq!( + routes[2], + ("/cluster-2".to_string(), "cluster-2".to_string()) + ); + } +} From 3838317371268188befdd03ac5f596c2cfdf60ad Mon Sep 17 00:00:00 2001 From: aryehlev Date: Sun, 2 Aug 2026 10:45:48 +0300 Subject: [PATCH 3/3] share one ProcessingDone signal per response. --- tonic-xds/src/xds/resource_manager.rs | 6 +- xds-client/src/client/watch.rs | 34 +-- xds-client/src/client/worker.rs | 305 +++++++++++++++++++++----- 3 files changed, 272 insertions(+), 73 deletions(-) diff --git a/tonic-xds/src/xds/resource_manager.rs b/tonic-xds/src/xds/resource_manager.rs index 98543cc10..9f0c80f1d 100644 --- a/tonic-xds/src/xds/resource_manager.rs +++ b/tonic-xds/src/xds/resource_manager.rs @@ -384,21 +384,21 @@ mod tests { fn ok_event(resource: Arc) -> ResourceEvent { ResourceEvent::ResourceChanged { result: Ok(resource), - done: ProcessingDone::noop(), + done: ProcessingDone::detached(), } } fn err_event() -> ResourceEvent { ResourceEvent::ResourceChanged { result: Err(xds_client::Error::ResourceDoesNotExist), - done: ProcessingDone::noop(), + done: ProcessingDone::detached(), } } fn ambient_event() -> ResourceEvent { ResourceEvent::AmbientError { error: xds_client::Error::ResourceDoesNotExist, - done: ProcessingDone::noop(), + done: ProcessingDone::detached(), } } diff --git a/xds-client/src/client/watch.rs b/xds-client/src/client/watch.rs index cb553dceb..5b10069b1 100644 --- a/xds-client/src/client/watch.rs +++ b/xds-client/src/client/watch.rs @@ -70,33 +70,33 @@ use crate::resource::{DecodedResource, Resource}; /// } /// ``` #[derive(Debug)] -pub struct ProcessingDone(Option>); +pub struct ProcessingDone(Option>>); impl ProcessingDone { - /// Create a channel pair for signaling. + /// Create the shared signal for one response. /// - /// Returns the `ProcessingDone` sender and a receiver future that resolves - /// when the sender is dropped. + /// Returns a `ProcessingDone` token and a receiver future that resolves + /// once the token and every [`share`](Self::share) of it are dropped + /// (dropping the last `Arc` drops the sender, which wakes the receiver). pub(crate) fn channel() -> (Self, oneshot::Receiver<()>) { let (tx, rx) = oneshot::channel(); - (Self(Some(tx)), rx) + (Self(Some(Arc::new(tx))), rx) } - /// Creates a no-op token that signals nothing when dropped. + /// Another token tied to the same response's signal. /// - /// Requires the `test-util` feature. - #[cfg(feature = "test-util")] - pub fn noop() -> Self { - Self(None) + /// All tokens for one response share a single allocation; the receiver + /// resolves only after every one of them is dropped. + pub(crate) fn share(&self) -> Self { + Self(self.0.clone()) } -} -impl Drop for ProcessingDone { - fn drop(&mut self) { - // Auto-signal on drop to prevent deadlocks. - if let Some(tx) = self.0.take() { - let _ = tx.send(()); - } + /// A token that signals nothing when dropped. + /// + /// Used internally for events that do not participate in ADS flow + /// control, and useful for constructing [`ResourceEvent`]s in tests. + pub fn detached() -> Self { + Self(None) } } diff --git a/xds-client/src/client/worker.rs b/xds-client/src/client/worker.rs index eafa3b3a6..ba6a8cb6a 100644 --- a/xds-client/src/client/worker.rs +++ b/xds-client/src/client/worker.rs @@ -377,7 +377,8 @@ impl CachedResource { /// Convert cached state to a ResourceEvent for notifying watchers. /// Returns None if state is Requested (nothing to notify yet). fn to_event(&self) -> Option> { - let (done, _rx) = ProcessingDone::channel(); + // Cache-dump events for new watchers do not gate flow control. + let done = ProcessingDone::detached(); match &self.state { ResourceState::Received => { self.resource @@ -576,13 +577,12 @@ pub(crate) struct AdsWorker { } /// A watcher notification staged during response handling: the channel to -/// deliver on, the event (carrying its `ProcessingDone` token), and — for -/// events that participate in ADS flow control — the receiver that resolves -/// when the watcher finishes processing. +/// deliver on and the event, carrying its `ProcessingDone` token. Events +/// that participate in ADS flow control share the response's single +/// `ProcessingDone` signal; the others carry a detached token. type Delivery = ( mpsc::Sender>, ResourceEvent, - Option>, ); /// In-flight watcher deliveries for one response: sends every staged event @@ -795,7 +795,7 @@ where Ok(Some(bytes)) => { saw_response = true; match self.handle_response(&mut stream, bytes).await { - Ok(deliveries) => pending = Self::dispatch_pending(deliveries), + Ok(dispatch) => pending = dispatch, Err(_) => return ConnectedOutcome::Failed { saw_response }, } } @@ -825,24 +825,22 @@ where /// Build the in-flight delivery future for one response's staged watcher /// notifications, or `None` when there is nothing to deliver. - fn dispatch_pending(deliveries: Vec) -> Option { + fn dispatch_pending( + deliveries: Vec, + done_rx: oneshot::Receiver<()>, + ) -> Option { if deliveries.is_empty() { return None; } Some(Box::pin(async move { - let mut done_rxs = Vec::with_capacity(deliveries.len()); - for (event_tx, event, done_rx) in deliveries { + for (event_tx, event) in deliveries { // Backpressure: await if the watcher's channel is full. Send // errors (watcher dropped) are ignored; the rejected event's - // `ProcessingDone` token resolves its receiver on drop. + // `ProcessingDone` token drops with it. let _ = event_tx.send(event).await; - if let Some(rx) = done_rx { - done_rxs.push(rx); - } - } - for rx in done_rxs { - let _ = rx.await; } + // Resolves once every token sharing this response's signal drops. + let _ = done_rx.await; })) } @@ -1031,25 +1029,31 @@ where /// - Missing resources (for types with ALL_RESOURCES_REQUIRED_IN_SOTW) are marked deleted /// /// Cache/state updates and the ACK/NACK happen here; the watcher - /// notifications are only *staged* and returned as [`Delivery`]s for - /// `run_connected` to drive, so a slow (or stuck) watcher delays reading - /// the next response — ADS flow control — without freezing command - /// processing. + /// notifications are only *staged* and returned as a [`PendingDispatch`] + /// future (`None` when there is nothing to deliver) for `run_connected` + /// to drive, so a slow (or stuck) watcher delays reading the next + /// response — ADS flow control — without freezing command processing. async fn handle_response( &mut self, stream: &mut S, bytes: Bytes, - ) -> Result> { + ) -> Result> { let response = self.codec.decode_response(bytes)?; let type_url = response.type_url.clone(); let (type_url_arc, decoder) = match self.type_states.get(&type_url) { Some(s) => (Arc::clone(&s.type_url), &s.decoder), None => { - return Ok(Vec::new()); + return Ok(None); } }; + // One shared `ProcessingDone` signal for the whole response: every + // flow-control event carries a share of it, and the receiver resolves + // when the last share (including the original, dropped on return + // from this function) is gone. + let (done, done_rx) = ProcessingDone::channel(); + // Decode all resources, tracking valid and invalid separately. // Per A46, we accept valid resources even if some fail validation. // Per A88, we categorize errors: @@ -1096,7 +1100,7 @@ where // reflects the accepted config regardless of watcher progress. let mut deliveries = Vec::new(); - self.dispatch_resources(&mut deliveries, &type_url, valid_resources); + self.dispatch_resources(&mut deliveries, &type_url, valid_resources, &done); // Only notify watchers for per-resource errors (where we know the name). // Top-level errors have no associated name, so no watcher to notify. @@ -1107,7 +1111,7 @@ where // Detect deleted resources (per A53): // For resource types with ALL_RESOURCES_REQUIRED_IN_SOTW = true, // any previously-received resource not in this response is deleted. - self.detect_deleted_resources(&mut deliveries, &type_url, &received_names); + self.detect_deleted_resources(&mut deliveries, &type_url, &received_names, &done); let has_errors = !top_level_errors.is_empty() || !per_resource_errors.is_empty(); if !has_errors { @@ -1138,19 +1142,21 @@ where .await?; } - Ok(deliveries) + Ok(Self::dispatch_pending(deliveries, done_rx)) } /// Update the cache from decoded resources and stage watcher deliveries. /// - /// The staged events carry `ProcessingDone` receivers that gate reading - /// the next response (ADS flow control); the sends themselves happen in - /// the [`PendingDispatch`] future, with backpressure on full channels. + /// The staged events share the response's `ProcessingDone` signal, which + /// gates reading the next response (ADS flow control); the sends + /// themselves happen in the [`PendingDispatch`] future, with backpressure + /// on full channels. fn dispatch_resources( &mut self, deliveries: &mut Vec, type_url: &str, resources: Vec, + done: &ProcessingDone, ) { let watcher_info: Vec<_> = match self.type_states.get_mut(type_url) { Some(s) => { @@ -1181,14 +1187,13 @@ where let resource_name = resource.name().to_string(); let resource = Arc::new(resource); - for (_watcher_id, event_tx, subscription) in watcher_info.clone() { + for (_watcher_id, event_tx, subscription) in &watcher_info { if subscription.matches(&resource_name) { - let (done, rx) = ProcessingDone::channel(); let event = ResourceEvent::ResourceChanged { result: Ok(Arc::clone(&resource)), - done, + done: done.share(), }; - deliveries.push((event_tx, event, Some(rx))); + deliveries.push((event_tx.clone(), event)); } } } @@ -1198,7 +1203,7 @@ where /// /// Per gRFC A46/A88, errors are routed only to watchers interested in /// that specific resource (plus wildcard watchers). Error events do not - /// gate flow control (no `ProcessingDone` receiver is staged). + /// gate flow control (they carry a detached `ProcessingDone` token). fn notify_resource_error( &mut self, deliveries: &mut Vec, @@ -1224,12 +1229,11 @@ where .remove(&(type_url.to_string(), resource_name.to_string())); for event_tx in type_state.matching_watchers(resource_name) { - let (done, _rx) = ProcessingDone::channel(); let event = ResourceEvent::ResourceChanged { result: Err(Error::Validation(error.to_string())), - done, + done: ProcessingDone::detached(), }; - deliveries.push((event_tx, event, None)); + deliveries.push((event_tx, event)); } } @@ -1244,6 +1248,7 @@ where deliveries: &mut Vec, type_url: &str, received_names: &HashSet, + done: &ProcessingDone, ) { let type_state = match self.type_states.get_mut(type_url) { Some(s) => s, @@ -1269,12 +1274,11 @@ where .insert(name.clone(), CachedResource::does_not_exist()); for event_tx in type_state.matching_watchers(&name) { - let (done, rx) = ProcessingDone::channel(); let event = ResourceEvent::ResourceChanged { result: Err(Error::ResourceDoesNotExist), - done, + done: done.share(), }; - deliveries.push((event_tx, event, Some(rx))); + deliveries.push((event_tx, event)); } } @@ -1404,10 +1408,9 @@ where .sync_resource_counts(&type_state.type_url, &counts); for event_tx in type_state.matching_watchers(name) { - let (done, _rx) = ProcessingDone::channel(); let event = ResourceEvent::ResourceChanged { result: Err(Error::ResourceDoesNotExist), - done, + done: ProcessingDone::detached(), }; let _ = event_tx.send(event).await; } @@ -1623,7 +1626,7 @@ mod flow_control_tests { use bytes::Bytes; use crate::client::config::ClientConfig; - use crate::client::watch::ResourceEvent; + use crate::client::watch::{ResourceEvent, ResourceWatcher}; use crate::codec::XdsCodec; use crate::error::Result; use crate::message::{DiscoveryRequest, DiscoveryResponse, Node, ResourceAny}; @@ -1633,8 +1636,10 @@ mod flow_control_tests { use crate::{XdsClient, error::Error}; const TEST_TYPE_URL: &str = "type.googleapis.com/test.Resource"; + const SOTW_TYPE_URL: &str = "type.googleapis.com/test.SotwResource"; - /// Minimal resource: the message is the resource name itself. + /// Minimal resource: the message is the resource name itself. Names + /// starting with `bad` fail validation (a per-resource error, per A46). #[derive(Debug, Clone)] struct TestResource; @@ -1651,6 +1656,32 @@ mod flow_control_tests { message } + fn validate(message: Self::Message) -> Result { + if message.starts_with("bad") { + return Err(Error::Validation("bad resource".to_string())); + } + Ok(Self) + } + } + + /// Like [`TestResource`] but with `ALL_RESOURCES_REQUIRED_IN_SOTW`, so + /// resources missing from a response are treated as deleted (gRFC A53). + #[derive(Debug, Clone)] + struct SotwResource; + + impl Resource for SotwResource { + type Message = String; + const TYPE_URL: TypeUrl = TypeUrl::new(SOTW_TYPE_URL); + const ALL_RESOURCES_REQUIRED_IN_SOTW: bool = true; + + fn deserialize(bytes: Bytes) -> Result { + String::from_utf8(bytes.to_vec()).map_err(|e| Error::Validation(e.to_string())) + } + + fn name(message: &Self::Message) -> &str { + message + } + fn validate(_message: Self::Message) -> Result { Ok(Self) } @@ -1696,13 +1727,21 @@ mod flow_control_tests { } } - fn response(version: &str, nonce: &str, names: &[&str]) -> Bytes { + fn response_for(type_url: &str, version: &str, nonce: &str, names: &[&str]) -> Bytes { Bytes::from(format!( - "{TEST_TYPE_URL}\n{version}\n{nonce}\n{}", + "{type_url}\n{version}\n{nonce}\n{}", names.join(",") )) } + fn response(version: &str, nonce: &str, names: &[&str]) -> Bytes { + response_for(TEST_TYPE_URL, version, nonce, names) + } + + fn sotw_response(version: &str, nonce: &str, names: &[&str]) -> Bytes { + response_for(SOTW_TYPE_URL, version, nonce, names) + } + /// (version_info, response_nonce) of an encoded request. fn parse_request(bytes: &Bytes) -> (String, String) { let text = String::from_utf8(bytes.to_vec()).unwrap(); @@ -1713,18 +1752,14 @@ mod flow_control_tests { (version, nonce) } - /// Client watching `res-0` with an established mock stream, its initial - /// request already drained. - async fn connected_client() -> ( - XdsClient, - crate::client::watch::ResourceWatcher, - MockServer, - ) { + /// Client watching `res-0` (of resource type `T`) with an established + /// mock stream, its initial request already drained. + async fn connected_client_for() -> (XdsClient, ResourceWatcher, MockServer) { let (builder, mut servers) = mock_transport(); let config = ClientConfig::new(Node::new("test", "0"), "mock:///xds"); let client = XdsClient::builder(config, builder, FakeCodec, TokioRuntime).build(); - let watcher = client.watch::("res-0").await; + let watcher = client.watch::("res-0").await; let mut server = tokio::time::timeout(Duration::from_secs(5), servers.recv()) .await .expect("timed out waiting for stream") @@ -1736,6 +1771,52 @@ mod flow_control_tests { (client, watcher, server) } + async fn connected_client() -> (XdsClient, ResourceWatcher, MockServer) { + connected_client_for::().await + } + + /// Watch `name` and wait for the resulting subscription request, so the + /// watcher is registered before the test sends a response. + async fn watch_synced( + client: &XdsClient, + server: &mut MockServer, + name: &str, + ) -> ResourceWatcher { + let watcher = client.watch::(name).await; + let _request = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("timed out waiting for subscription request") + .expect("stream closed"); + watcher + } + + /// Next event, unwrapped to its result and `ProcessingDone` token. + async fn next_changed( + watcher: &mut ResourceWatcher, + ) -> ( + Result>, + crate::client::watch::ProcessingDone, + ) { + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for event") + .expect("watcher closed"); + match event { + ResourceEvent::ResourceChanged { result, done } => (result, done), + ResourceEvent::AmbientError { .. } => panic!("unexpected ambient error"), + } + } + + /// Asserts no event is delivered to `watcher` within a short window. + async fn assert_no_event(watcher: &mut ResourceWatcher, message: &str) { + assert!( + tokio::time::timeout(Duration::from_millis(200), watcher.next()) + .await + .is_err(), + "{message}" + ); + } + /// A watcher that issues more commands than the command channel buffers /// while holding its `ProcessingDone` token must not deadlock the worker. /// @@ -1846,4 +1927,122 @@ mod flow_control_tests { .expect("stream closed"); assert_eq!(parse_request(&ack), ("2".to_string(), "n2".to_string())); } + + /// The next response is gated until *every* watcher drops its + /// `ProcessingDone` token, not just the first one. + #[tokio::test] + async fn next_response_gated_until_all_watchers_signal() { + let (client, mut w1, mut server) = connected_client().await; + let mut w2 = watch_synced(&client, &mut server, "res-1").await; + + server + .responses + .send(Ok(Some(response("1", "n1", &["res-0", "res-1"])))) + .unwrap(); + let (r1, done1) = next_changed(&mut w1).await; + let (r2, done2) = next_changed(&mut w2).await; + assert!(r1.is_ok() && r2.is_ok()); + + drop(done1); + server + .responses + .send(Ok(Some(response("2", "n2", &["res-0", "res-1"])))) + .unwrap(); + assert_no_event(&mut w1, "second response delivered while a token was held").await; + + drop(done2); + assert!(next_changed(&mut w1).await.0.is_ok()); + assert!(next_changed(&mut w2).await.0.is_ok()); + } + + /// Validation-error events do not gate flow control (gRFC A46/A88): + /// the response is NACKed, valid resources are still delivered, and a + /// held error token must not delay the next response. + #[tokio::test] + async fn error_events_do_not_gate_next_response() { + let (client, mut w_ok, mut server) = connected_client().await; + let mut w_bad = watch_synced(&client, &mut server, "bad-0").await; + + server + .responses + .send(Ok(Some(response("1", "n1", &["res-0", "bad-0"])))) + .unwrap(); + let (result, done_ok) = next_changed(&mut w_ok).await; + assert!(result.is_ok()); + let (result, _err_done) = next_changed(&mut w_bad).await; + assert!(matches!(result, Err(Error::Validation(_)))); + + // NACK keeps the old (empty) version. + let nack = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("NACK not sent") + .expect("stream closed"); + assert_eq!(parse_request(&nack), ("".to_string(), "n1".to_string())); + + drop(done_ok); + server + .responses + .send(Ok(Some(response("2", "n2", &["res-0"])))) + .unwrap(); + // `_err_done` is still held; it must not gate this delivery. + assert!(next_changed(&mut w_ok).await.0.is_ok()); + } + + /// Deletion events (SotW resource missing from a response, gRFC A53) + /// gate the next response like regular updates. + #[tokio::test] + async fn deletion_events_gate_next_response() { + let (_client, mut watcher, server) = connected_client_for::().await; + + server + .responses + .send(Ok(Some(sotw_response("1", "n1", &["res-0"])))) + .unwrap(); + let (result, done) = next_changed(&mut watcher).await; + assert!(result.is_ok()); + drop(done); + + // res-0 missing from the SotW response: deleted. + server + .responses + .send(Ok(Some(sotw_response("2", "n2", &[])))) + .unwrap(); + let (result, deletion_done) = next_changed(&mut watcher).await; + assert!(matches!(result, Err(Error::ResourceDoesNotExist))); + + server + .responses + .send(Ok(Some(sotw_response("3", "n3", &["res-0"])))) + .unwrap(); + assert_no_event( + &mut watcher, + "response delivered while the deletion token was held", + ) + .await; + + drop(deletion_done); + assert!(next_changed(&mut watcher).await.0.is_ok()); + } + + /// Events staged for a watcher that was dropped must not wedge flow + /// control: their failed sends release the shared signal. + #[tokio::test] + async fn dropped_watcher_does_not_stall_flow_control() { + let (client, mut w1, mut server) = connected_client().await; + drop(watch_synced(&client, &mut server, "res-1").await); + + server + .responses + .send(Ok(Some(response("1", "n1", &["res-0", "res-1"])))) + .unwrap(); + let (result, done) = next_changed(&mut w1).await; + assert!(result.is_ok()); + drop(done); + + server + .responses + .send(Ok(Some(response("2", "n2", &["res-0", "res-1"])))) + .unwrap(); + assert!(next_changed(&mut w1).await.0.is_ok()); + } }