diff --git a/rs/xnet/payload_builder/src/lib.rs b/rs/xnet/payload_builder/src/lib.rs index 5ed7ecd4d63e..e4813c250f23 100644 --- a/rs/xnet/payload_builder/src/lib.rs +++ b/rs/xnet/payload_builder/src/lib.rs @@ -22,7 +22,7 @@ use http_body_util::BodyExt; use hyper::{Request, StatusCode, Uri}; use hyper_util::client::legacy::Client; use hyper_util::rt::{TokioExecutor, TokioTimer}; -use ic_config::message_routing::MAX_STREAM_MESSAGES; +use ic_config::message_routing::{ADVERT_MAX_BODY_BYTES, MAX_STREAM_MESSAGES}; use ic_crypto_tls_interfaces::TlsConfig; use ic_interfaces::crypto::ErrorReproducibility; use ic_interfaces::messaging::{ @@ -1987,6 +1987,17 @@ pub trait XNetClient: Sync + Send { &self, endpoint: &EndpointLocator, ) -> Result; + + /// Posts an advert — our own certified stream header — to the given + /// `XNetEndpoint`. + /// + /// On success, returns the peer's response: its certified header iff it already + /// reflects the contents of the advert, else `None`. + async fn post_advert( + &self, + endpoint: &EndpointLocator, + advert: CertifiedStreamSlice, + ) -> Result, XNetClientError>; } type XNetRequestBody = http_body_util::Full; @@ -1995,7 +2006,7 @@ type XNetRequestBody = http_body_util::Full; /// configuration and connection pooling). struct XNetClientImpl { /// An HTTP client to be used for querying. - http_client: Client>, + http_client: Client, /// Response body (encoded slice) size. response_body_size: HistogramVec, @@ -2032,7 +2043,7 @@ impl XNetClientImpl { // query against it timing out. With keep-alive, such a connection is closed // within `interval + timeout` seconds and a fresh connection is established // on the next query. - let http_client: Client> = + let http_client: Client = Client::builder(TokioExecutor::new()) .http2_only(true) // Timer required by HTTP/2 keep-alive. @@ -2125,6 +2136,69 @@ impl XNetClientImpl { )), } } + + async fn post_advert_impl( + &self, + endpoint: &EndpointLocator, + advert: CertifiedStreamSlice, + ) -> Result, XNetClientError> { + let body = pb::CertifiedStreamSlice::proxy_encode(advert); + // The receiver refuses anything larger, so there is no point in sending it. + if body.len() > ADVERT_MAX_BODY_BYTES { + return Err(XNetClientError::AdvertTooLarge(body.len())); + } + let request = Request::post(endpoint.url.clone()) + .header(hyper::header::CONTENT_TYPE, "application/x-protobuf") + .body(XNetRequestBody::from(body)) + .expect("failed to build advert request"); + + // TODO(MR-28) Make timeout configurable. + let result = tokio::time::timeout(Duration::from_secs(5), async { + let response = self + .http_client + .request(request) + .await + .map_err(XNetClientError::RequestFailed)?; + + let status = response.status(); + + // A reply is a header-only slice, just like the advert. + let content = http_body_util::Limited::new(response.into_body(), ADVERT_MAX_BODY_BYTES) + .collect() + .await + .map(|collected| collected.to_bytes()) + .map_err(XNetClientError::BodyReadError)?; + + Ok((status, content)) + }) + .await; + + let (status, bytes) = result.map_err(|_| XNetClientError::Timeout)??; + + match status { + // Advert accepted. + StatusCode::NO_CONTENT => Ok(None), + + // Everything in our advert was already inducted and certified: the reply is a + // certified header that proves it. + StatusCode::OK => pb::CertifiedStreamSlice::proxy_decode(bytes.as_ref()) + .map(Some) + .map_err(XNetClientError::ProxyDecodeError), + + _ => Err(XNetClientError::ErrorResponse( + status, + String::from_utf8_lossy(bytes.as_ref()).to_string(), + )), + } + } + + /// Updates the node's health status, based on whether it served the request. + fn update_node_health(&self, node_id: NodeId, result: &Result) { + match result { + Err(e) if e.is_node_failure() => self.unhealthy_nodes.observe_failure(node_id), + _ => self.unhealthy_nodes.observe_success(node_id), + } + } } #[async_trait] @@ -2134,14 +2208,17 @@ impl XNetClient for XNetClientImpl { endpoint: &EndpointLocator, ) -> Result { let result = self.query_impl(endpoint).await; + self.update_node_health(endpoint.node_id, &result); + result + } - // Record whether the node served the request, so that node selection - // can skip the ones that don't. - match &result { - Err(e) if e.is_node_failure() => self.unhealthy_nodes.observe_failure(endpoint.node_id), - _ => self.unhealthy_nodes.observe_success(endpoint.node_id), - } - + async fn post_advert( + &self, + endpoint: &EndpointLocator, + advert: CertifiedStreamSlice, + ) -> Result, XNetClientError> { + let result = self.post_advert_impl(endpoint, advert).await; + self.update_node_health(endpoint.node_id, &result); result } } @@ -2160,6 +2237,8 @@ pub enum XNetClientError { BodyReadError(Box), #[error("Error decoding XNet proto into Rust struct: {0}")] ProxyDecodeError(ProxyDecodeError), + #[error("Advert of {0} bytes exceeds the {ADVERT_MAX_BODY_BYTES} byte limit")] + AdvertTooLarge(usize), } impl XNetClientError { @@ -2186,6 +2265,9 @@ impl XNetClientError { // Definitely not an error, there's merely no new content. XNetClientError::NoContent => false, + + // Our own doing, nothing to do with the node. + XNetClientError::AdvertTooLarge(..) => false, } } @@ -2198,6 +2280,7 @@ impl XNetClientError { XNetClientError::ErrorResponse(status, _) => format!("HTTP_{}", status.as_u16()), XNetClientError::BodyReadError(..) => "BodyReadError".to_string(), XNetClientError::ProxyDecodeError(..) => STATUS_DECODE_ERROR.to_string(), + XNetClientError::AdvertTooLarge(..) => "AdvertTooLarge".to_string(), } } } diff --git a/rs/xnet/payload_builder/src/xnet_client_tests.rs b/rs/xnet/payload_builder/src/xnet_client_tests.rs index a92112a6e053..a4306f456357 100644 --- a/rs/xnet/payload_builder/src/xnet_client_tests.rs +++ b/rs/xnet/payload_builder/src/xnet_client_tests.rs @@ -8,7 +8,7 @@ use axum::{ Router, http::{HeaderMap, StatusCode, header::CONTENT_TYPE}, response::IntoResponse, - routing::{MethodRouter, get}, + routing::{MethodRouter, get, post}, }; use hyper::Uri; use ic_crypto_tls_interfaces_mocks::MockTlsConfig; @@ -329,6 +329,114 @@ async fn query_request_failed() { assert_eq!(Some(1), fetch_int_gauge(metrics, METRIC_UNHEALTHY_NODES)); } +/// An advert that the peer answers with its own certified header, because it +/// brought it nothing new. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn advert_reply() { + let metrics = MetricsRegistry::new(); + let reply = get_stream_slice_for_testing(); + let expected = reply.clone(); + + let respond_with_reply = + post(move || proto_axum_response::<_, pb::CertifiedStreamSlice>(reply.clone())); + + let result = with_test_replica_logger(|log| async { + do_xnet_client_advert(&make_xnet_client(&metrics, log), respond_with_reply).await + }) + .await; + + assert_eq!(Some(expected), result.unwrap()); + assert_eq!(Some(0), fetch_int_gauge(&metrics, METRIC_UNHEALTHY_NODES)); +} + +/// An advert the peer took, with nothing to reply. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn advert_no_reply() { + let metrics = MetricsRegistry::new(); + + let respond_with_no_content = post(|| async { StatusCode::NO_CONTENT }); + + let result = with_test_replica_logger(|log| async { + do_xnet_client_advert(&make_xnet_client(&metrics, log), respond_with_no_content).await + }) + .await; + + assert_eq!(None, result.unwrap()); + assert_eq!(Some(0), fetch_int_gauge(&metrics, METRIC_UNHEALTHY_NODES)); +} + +/// Being refused (e.g. rate limited) says nothing about the node's health. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn advert_refused() { + let metrics = MetricsRegistry::new(); + + let respond_with_too_many_requests = + post(|| async { (StatusCode::TOO_MANY_REQUESTS, b"Slow down".to_vec()) }); + + let result = with_test_replica_logger(|log| async { + do_xnet_client_advert( + &make_xnet_client(&metrics, log), + respond_with_too_many_requests, + ) + .await + }) + .await; + + match result { + Err(XNetClientError::ErrorResponse(StatusCode::TOO_MANY_REQUESTS, _)) => (), + _ => panic!("Expecting Err(ErrorResponse(_)), got {result:?}"), + } + assert_eq!(Some(0), fetch_int_gauge(&metrics, METRIC_UNHEALTHY_NODES)); +} + +/// A node that fails to serve an advert is marked unhealthy, so that node +/// selection skips it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn advert_server_error() { + let metrics = MetricsRegistry::new(); + + let respond_with_server_error = + post(|| async { (StatusCode::SERVICE_UNAVAILABLE, b"Oops".to_vec()) }); + + let result = with_test_replica_logger(|log| async { + do_xnet_client_advert(&make_xnet_client(&metrics, log), respond_with_server_error).await + }) + .await; + + match result { + Err(XNetClientError::ErrorResponse(StatusCode::SERVICE_UNAVAILABLE, _)) => (), + _ => panic!("Expecting Err(ErrorResponse(_)), got {result:?}"), + } + assert_eq!(Some(1), fetch_int_gauge(&metrics, METRIC_UNHEALTHY_NODES)); +} + +/// An advert above `ADVERT_MAX_BODY_BYTES` is not sent at all: the peer would +/// refuse it, and the fault is ours. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn advert_too_large() { + let metrics = MetricsRegistry::new(); + let oversized = CertifiedStreamSlice { + payload: vec![0; ADVERT_MAX_BODY_BYTES], + ..get_stream_slice_for_testing() + }; + + // Nothing is listening: an advert actually sent would fail to connect. + let nowhere = SocketAddr::from(([127, 0, 0, 1], 1)); + + let result = with_test_replica_logger(|log| async { + make_xnet_client(&metrics, log) + .post_advert(&advert_endpoint(nowhere), oversized) + .await + }) + .await; + + match result { + Err(XNetClientError::AdvertTooLarge(bytes)) => assert!(bytes > ADVERT_MAX_BODY_BYTES), + _ => panic!("Expecting Err(AdvertTooLarge(_)), got {result:?}"), + } + assert_eq!(Some(0), fetch_int_gauge(&metrics, METRIC_UNHEALTHY_NODES)); +} + /// Returns the result of invoking `xnet_client.query()` against an HTTP server /// in a spawned thread that processes a single request using `handle_request`. async fn do_xnet_client_query( @@ -358,6 +466,29 @@ async fn do_async_query( xnet_client.query(&endpoint).await } +/// Serves `method_router` and posts an advert against it. +async fn do_xnet_client_advert( + xnet_client: &XNetClientImpl, + method_router: MethodRouter, +) -> Result, XNetClientError> { + let router = Router::new().route("/", method_router); + let socket = start_server(router).await; + + xnet_client + .post_advert(&advert_endpoint(socket), get_stream_slice_for_testing()) + .await +} + +fn advert_endpoint(socket: SocketAddr) -> EndpointLocator { + EndpointLocator { + node_id: LOCAL_NODE, + url: format!("http://aaaaa-aa.1@{}:{}", socket.ip(), socket.port()) + .parse::() + .unwrap(), + proximity: PeerLocation::Local, + } +} + async fn start_server(router: Router) -> SocketAddr { let address = SocketAddr::from(([127, 0, 0, 1], 0)); let listener = tokio::net::TcpListener::bind(address).await.unwrap(); @@ -368,6 +499,7 @@ async fn start_server(router: Router) -> SocketAddr { }); socket } + /// Generates a stream slice from `DST_SUBNET`. fn get_stream_slice_for_testing() -> CertifiedStreamSlice { make_certified_stream_slice( diff --git a/rs/xnet/payload_builder/tests/xnet_payload_builder.rs b/rs/xnet/payload_builder/tests/xnet_payload_builder.rs index 760e466806e0..c7c6bce9a958 100644 --- a/rs/xnet/payload_builder/tests/xnet_payload_builder.rs +++ b/rs/xnet/payload_builder/tests/xnet_payload_builder.rs @@ -1008,6 +1008,14 @@ impl XNetClient for FakeXNetClient { FakeXNetClientError::NoContent => XNetClientError::NoContent, }) } + + async fn post_advert( + &self, + _endpoint: &EndpointLocator, + _advert: CertifiedStreamSlice, + ) -> Result, XNetClientError> { + unimplemented!("no advert sending in these tests") + } } /// A replacement for `XNetClientError` because `XNetClientError` is not `Clone`