Merge pull request #5 from dfinity/igornovg/cli-v2

feat(BOUN-1233): prerequisite for CLI improvement in services

.gitignore

@@ -1,2 +1,3 @@
+.idea
 /target
 Cargo.lock

Cargo.toml

@@ -8,85 +8,89 @@ repository = "https://github.com/dfinity/ic-bn-lib"
 readme = "README.md"
 
 [dependencies]
-ahash = "0.8"
-anyhow = "1.0"
-arc-swap = "1"
-async-trait = "0.1.81"
-axum = "0.7"
-backoff = { version = "0.4", features = ["tokio"] }
-base64 = "0.22"
-bytes = "1.6"
-clap = { version = "4.5", features = ["derive", "string", "env"] }
-clap_derive = "4.5"
-chacha20poly1305 = "0.10"
+ahash = "0.8.11"
+anyhow = "1.0.92"
+arc-swap = "1.7.1"
+async-trait = "0.1.83"
+axum = "0.7.7"
+backoff = { version = "0.4.0", features = ["tokio"] }
+base64 = "0.22.1"
+bytes = "1.8.0"
+clap = { version = "4.5.20", features = ["derive", "string", "env"] }
+clap_derive = "4.5.18"
+chacha20poly1305 = "0.10.1"
 cloudflare = { git = "https://github.com/cloudflare/cloudflare-rs.git", rev = "f14720e42184ee176a97676e85ef2d2d85bc3aae", default-features = false, features = [
     "rustls-tls",
 ] }
-derive-new = "0.6"
-fqdn = "0.3"
-futures = "0.3"
-futures-util = "0.3"
-hickory-proto = "0.24"
-hickory-resolver = { version = "0.24", features = [
+derive-new = "0.7.0"
+fqdn = "0.4.1"
+futures = "0.3.31"
+futures-util = "0.3.31"
+hickory-proto = "0.24.1"
+hickory-resolver = { version = "0.24.1", features = [
     "dns-over-https-rustls",
     "webpki-roots",
     "dnssec-ring",
 ] }
-http = "1.1"
-http-body = "1.0"
-http-body-util = "0.1"
-humantime = "2.1"
-hyper = "1.5"
-hyper-util = { version = "0.1", features = ["full"] }
-instant-acme = { version = "0.7.1", default-features = false, features = [
+http = "1.1.0"
+http-body = "1.0.1"
+http-body-util = "0.1.2"
+humantime = "2.1.0"
+hyper = "1.5.0"
+hyper-util = { version = "0.1.10", features = ["full"] }
+instant-acme = { version = "0.7.2", default-features = false, features = [
     "ring",
     "hyper-rustls",
 ] }
-mockall = "0.12"
-moka = { version = "0.12", features = ["sync", "future"] }
-prometheus = "0.13"
-prost = "0.13"
-prost-types = "0.13"
-rand = "0.8"
+mockall = "0.13.0"
+moka = { version = "0.12.8", features = ["sync", "future"] }
+parse-size = { version = "1.1.0", features = ["std"] }
+prometheus = "0.13.4"
+prost = "0.13.3"
+prost-types = "0.13.3"
+rand = "0.8.5"
 rcgen = "0.13.1"
-reqwest = { version = "0.12.7", default-features = false, features = [
+reqwest = { version = "0.12.9", default-features = false, features = [
     "http2",
     "rustls-tls",
     "hickory-dns",
     "json",
     "stream",
 ] }
-rustls = { version = "0.23.12", default-features = false, features = [
+rustls = { version = "0.23.16", default-features = false, features = [
     "ring",
     "std",
     "brotli",
 ] }
-rustls-acme = { version = "0.11", default-features = false, features = [
+rustls-acme = { version = "0.11.1", default-features = false, features = [
     "tls12",
     "ring",
 ] }
-rustls-pemfile = "2"
-serde_json = "1.0"
+rustls-pemfile = "2.2.0"
+rustls-platform-verifier = "0.4.0"
 scopeguard = "1.2.0"
-sha1 = "0.10"
-strum = { version = "0.26", features = ["derive"] }
-strum_macros = "0.26"
-sync_wrapper = "1.0"
+serde = "1.0.214"
+serde_json = "1.0.132"
+sha1 = "0.10.6"
+strum = { version = "0.26.3", features = ["derive"] }
+strum_macros = "0.26.4"
+sync_wrapper = "1.0.1"
 systemstat = "0.2.3"
-thiserror = "1.0"
-tokio = { version = "1.41", features = ["full"] }
-tokio-util = { version = "0.7", features = ["full"] }
+thiserror = "1.0.68"
+tokio = { version = "1.41.0", features = ["full"] }
+tokio-util = { version = "0.7.12", features = ["full"] }
 tokio-rustls = { version = "0.26.0", default-features = false, features = [
     "tls12",
     "logging",
     "ring",
 ] }
-tokio-io-timeout = "1.2"
-tower = { version = "0.5", features = ["util"] }
-tower-service = "0.3"
-tracing = "0.1"
-url = "2.5"
-uuid = { version = "1.10", features = ["v7"] }
+tokio-io-timeout = "1.2.0"
+tower = { version = "0.5.1", features = ["util"] }
+tower-service = "0.3.3"
+tracing = "0.1.40"
+url = "2.5.3"
+uuid = { version = "1.10.0", features = ["v7"] }
 vrl = { version = "0.19.0", default-features = false, features = ["value"] }
-x509-parser = "0.16"
-zeroize = { version = "1.8", features = ["derive"] }
+webpki-root-certs = "0.26.6"
+x509-parser = "0.16.0"
+zeroize = { version = "1.8.1", features = ["derive"] }

src/http/body.rs

@@ -73,7 +73,7 @@ impl Stream for SyncBodyDataStream {
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         loop {
             let mut pinned = pin!(self.inner.get_mut());
-            match futures_util::ready!(pinned.as_mut().poll_frame(cx)?) {
+            match ready!(pinned.as_mut().poll_frame(cx)?) {
                 Some(frame) => match frame.into_data() {
                     Ok(data) => return Poll::Ready(Some(Ok(data))),
                     Err(_frame) => {}
@@ -121,7 +121,7 @@ impl<D, E, S: Clone + Unpin> NotifyingBody<D, E, S> {
 impl<D, E, S: Clone + Unpin> HttpBody for NotifyingBody<D, E, S>
 where
     D: Buf,
-    E: std::string::ToString,
+    E: ToString,
 {
     type Data = D;
     type Error = E;
@@ -195,7 +195,7 @@ impl<D, E> CountingBody<D, E> {
 impl<D, E> HttpBody for CountingBody<D, E>
 where
     D: Buf,
-    E: std::string::ToString,
+    E: ToString,
 {
     type Data = D;
     type Error = E;
@@ -262,7 +262,7 @@ mod test {
         blahfoobarblahblah";
 
         let stream = tokio_util::io::ReaderStream::new(&data[..]);
-        let body = axum::body::Body::from_stream(stream);
+        let body = Body::from_stream(stream);
 
         let (body, rx) = CountingBody::new(body);
 
@@ -278,7 +278,7 @@ mod test {
     #[tokio::test]
     async fn test_counting_body_full() {
         let data = vec![0; 512];
-        let buf = bytes::Bytes::from_iter(data.clone());
+        let buf = Bytes::from_iter(data.clone());
         let body = http_body_util::Full::new(buf);
 
         let (body, rx) = CountingBody::new(body);
@@ -301,7 +301,7 @@ mod test {
         blahfoobarblahblah";
 
         let stream = tokio_util::io::ReaderStream::new(&data[..]);
-        let body = axum::body::Body::from_stream(stream);
+        let body = Body::from_stream(stream);
 
         let sig = 357;
         let (tx, mut rx) = mpsc::channel(10);

src/http/cache.rs

@@ -421,7 +421,7 @@ impl<K: KeyExtractor> Run for Cache<K> {
         interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
         loop {
-            tokio::select! {
+            select! {
                 biased;
 
                 () = token.cancelled() => {
@@ -871,7 +871,7 @@ mod tests {
             }
         }
 
-        assert!(refresh == 0);
+        assert_eq!(refresh, 0);
 
         // Check mid-expiration with high beta
         let now2 = now + Duration::from_secs(30);

src/http/client/cli.rs

@@ -0,0 +1,57 @@
+use std::time::Duration;
+
+use clap::Args;
+use humantime::parse_duration;
+
+use super::CloneableDnsResolver;
+
+#[derive(Args, Clone, Debug, Eq, PartialEq)]
+pub struct HttpClient {
+    /// Timeout for HTTP connection phase
+    #[clap(env, long, default_value = "5s", value_parser = parse_duration)]
+    pub http_client_timeout_connect: Duration,
+
+    /// Timeout for a single read request
+    #[clap(env, long, default_value = "15s", value_parser = parse_duration)]
+    pub http_client_timeout_read: Duration,
+
+    /// Timeout for the whole HTTP call: this includes connecting, sending request,
+    /// receiving response etc.
+    #[clap(env, long, default_value = "60s", value_parser = parse_duration)]
+    pub http_client_timeout: Duration,
+
+    /// How long to keep idle HTTP connections open
+    #[clap(env, long, default_value = "120s", value_parser = parse_duration)]
+    pub http_client_pool_idle: Duration,
+
+    /// TCP Keepalive interval
+    #[clap(env, long, default_value = "15s", value_parser = parse_duration)]
+    pub http_client_tcp_keepalive: Duration,
+
+    /// HTTP2 Keepalive interval
+    #[clap(env, long, default_value = "10s", value_parser = parse_duration)]
+    pub http_client_http2_keepalive: Duration,
+
+    /// HTTP2 Keepalive timeout
+    #[clap(env, long, default_value = "5s", value_parser = parse_duration)]
+    pub http_client_http2_keepalive_timeout: Duration,
+}
+
+impl<R: CloneableDnsResolver> From<&HttpClient> for super::Options<R> {
+    fn from(c: &HttpClient) -> Self {
+        Self {
+            timeout_connect: c.http_client_timeout_connect,
+            timeout_read: c.http_client_timeout_read,
+            timeout: c.http_client_timeout,
+            pool_idle_timeout: Some(c.http_client_pool_idle),
+            pool_idle_max: None,
+            tcp_keepalive: Some(c.http_client_tcp_keepalive),
+            http2_keepalive: Some(c.http_client_http2_keepalive),
+            http2_keepalive_timeout: c.http_client_http2_keepalive_timeout,
+            http2_keepalive_idle: false,
+            user_agent: "".into(),
+            tls_config: None,
+            dns_resolver: None,
+        }
+    }
+}

src/http/client.rs → src/http/client/mod.rs

@@ -1,3 +1,5 @@
+pub mod cli;
+
 use std::{
     fmt,
     sync::{
@@ -327,8 +329,8 @@ impl<G: GeneratesClients> Client for ReqwestClientDynamic<G> {
 
 pub fn basic_auth<U, P>(username: U, password: Option<P>) -> HeaderValue
 where
-    U: std::fmt::Display,
-    P: std::fmt::Display,
+    U: fmt::Display,
+    P: fmt::Display,
 {
     use base64::prelude::BASE64_STANDARD;
     use base64::write::EncoderWriter;
@@ -361,9 +363,9 @@ mod test {
     impl Client for TestClient {
         async fn execute(
             &self,
-            _req: reqwest::Request,
-        ) -> Result<reqwest::Response, reqwest::Error> {
-            let resp = ::http::Response::new(vec![]);
+            _req: Request,
+        ) -> Result<Response, reqwest::Error> {
+            let resp = http::Response::new(vec![]);
             tokio::time::sleep(Duration::from_millis(100)).await;
             Ok(resp.into())
         }
@@ -386,7 +388,7 @@ mod test {
         let mut futs = vec![];
         for _ in 0..200 {
             let req =
-                reqwest::Request::new(reqwest::Method::GET, url::Url::parse("http://foo").unwrap());
+                Request::new(reqwest::Method::GET, url::Url::parse("http://foo").unwrap());
 
             let cli = cli.clone();
             futs.push(async move { cli.execute(req).await });

src/http/headers.rs

@@ -25,7 +25,7 @@ pub const X_IC_CANISTER_ID: HeaderName = HeaderName::from_static("x-ic-canister-
 pub const X_IC_COUNTRY_CODE: HeaderName = HeaderName::from_static("x-ic-country-code");
 pub const X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id");
 pub const X_REQUESTED_WITH: HeaderName = HeaderName::from_static("x-requested-with");
-pub const X_REAL_IP: http::HeaderName = http::HeaderName::from_static("x-real-ip");
+pub const X_REAL_IP: HeaderName = HeaderName::from_static("x-real-ip");
 
 // Header values
 pub const CONTENT_TYPE_CBOR: HeaderValue = HeaderValue::from_static("application/cbor");

src/http/mod.rs

@@ -140,7 +140,7 @@ impl<T: AsyncRead + AsyncWrite + Send + Sync + Unpin> AsyncWrite for AsyncCounte
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &[u8],
-    ) -> Poll<std::io::Result<usize>> {
+    ) -> Poll<io::Result<usize>> {
         let poll = pin!(&mut self.inner).poll_write(cx, buf);
         if let Poll::Ready(Ok(v)) = &poll {
             self.stats.sent.fetch_add(*v as u64, Ordering::SeqCst);