feat: [DSM-148] XNetClient::post_advert()

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<CertifiedStreamSlice, XNetClientError>;
+
+    /// 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<Option<CertifiedStreamSlice>, XNetClientError>;
 }
 
 type XNetRequestBody = http_body_util::Full<hyper::body::Bytes>;
@@ -1995,7 +2006,7 @@ type XNetRequestBody = http_body_util::Full<hyper::body::Bytes>;
 /// configuration and connection pooling).
 struct XNetClientImpl {
     /// An HTTP client to be used for querying.
-    http_client: Client<TlsConnector, Request<XNetRequestBody>>,
+    http_client: Client<TlsConnector, XNetRequestBody>,
 
     /// 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<TlsConnector, Request<XNetRequestBody>> =
+        let http_client: Client<TlsConnector, XNetRequestBody> =
             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<Option<CertifiedStreamSlice>, 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<T>(&self, node_id: NodeId, result: &Result<T, XNetClientError>) {
+        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<CertifiedStreamSlice, XNetClientError> {
         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<Option<CertifiedStreamSlice>, 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<dyn std::error::Error + Send + Sync>),
     #[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(),
         }
     }
 }

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<Option<CertifiedStreamSlice>, 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::<Uri>()
+            .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(

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<Option<CertifiedStreamSlice>, XNetClientError> {
+        unimplemented!("no advert sending in these tests")
+    }
 }
 
 /// A replacement for `XNetClientError` because `XNetClientError` is not `Clone`