feat(sdk)!: provide request execution information

packages/rs-dapi-client/src/address_list.rs

@@ -1,6 +1,7 @@
 //! Subsystem to manage DAPI nodes.
 
 use chrono::Utc;
+use dapi_grpc::tonic::codegen::http;
 use dapi_grpc::tonic::transport::Uri;
 use rand::{rngs::SmallRng, seq::IteratorRandom, SeedableRng};
 use std::collections::HashSet;
@@ -20,6 +21,16 @@ pub struct Address {
     uri: Uri,
 }
 
+impl FromStr for Address {
+    type Err = AddressListError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Uri::from_str(s)
+            .map(Address::from)
+            .map_err(AddressListError::from)
+    }
+}
+
 impl PartialEq<Self> for Address {
     fn eq(&self, other: &Self) -> bool {
         self.uri == other.uri
@@ -81,6 +92,9 @@ impl Address {
 pub enum AddressListError {
     #[error("address {0} not found in the list")]
     AddressNotFound(#[cfg_attr(feature = "mocks", serde(with = "http_serde::uri"))] Uri),
+    #[error("unable parse address: {0}")]
+    #[cfg_attr(feature = "mocks", serde(skip))]
+    InvalidAddressUri(#[from] http::uri::InvalidUri),
 }
 
 /// A structure to manage DAPI addresses to select from
@@ -200,6 +214,7 @@ impl AddressList {
     }
 }
 
+// TODO: Must be changed to FromStr
 impl From<&str> for AddressList {
     fn from(value: &str) -> Self {
         let uri_list: Vec<Uri> = value

packages/rs-dapi-client/src/dapi_client.rs

@@ -1,9 +1,10 @@
 //! [DapiClient] definition.
 
-use backon::{ExponentialBuilder, Retryable};
+use backon::{ConstantBuilder, Retryable};
 use dapi_grpc::mock::Mockable;
 use dapi_grpc::tonic::async_trait;
 use std::fmt::Debug;
+use std::sync::atomic::AtomicUsize;
 use std::sync::{Arc, RwLock};
 use std::time::Duration;
 use tracing::Instrument;
@@ -12,19 +13,17 @@ use crate::address_list::AddressListError;
 use crate::connection_pool::ConnectionPool;
 use crate::{
     transport::{TransportClient, TransportRequest},
-    Address, AddressList, CanRetry, RequestSettings,
+    Address, AddressList, CanRetry, DapiRequestExecutor, ExecutionError, ExecutionResponse,
+    ExecutionResult, RequestSettings,
 };
 
 /// General DAPI request error type.
 #[derive(Debug, thiserror::Error)]
 #[cfg_attr(feature = "mocks", derive(serde::Serialize, serde::Deserialize))]
 pub enum DapiClientError<TE: Mockable> {
     /// The error happened on transport layer
-    #[error("transport error with {1}: {0}")]
-    Transport(
-        #[cfg_attr(feature = "mocks", serde(with = "dapi_grpc::mock::serde_mockable"))] TE,
-        Address,
-    ),
+    #[error("transport error: {0}")]
+    Transport(#[cfg_attr(feature = "mocks", serde(with = "dapi_grpc::mock::serde_mockable"))] TE),
     /// There are no valid DAPI addresses to use.
     #[error("no available addresses to use")]
     NoAvailableAddresses,
@@ -43,7 +42,7 @@ impl<TE: CanRetry + Mockable> CanRetry for DapiClientError<TE> {
         use DapiClientError::*;
         match self {
             NoAvailableAddresses => false,
-            Transport(transport_error, _) => transport_error.can_retry(),
+            Transport(transport_error) => transport_error.can_retry(),
             AddressList(_) => false,
             #[cfg(feature = "mocks")]
             Mock(_) => false,
@@ -73,21 +72,6 @@ impl<TE: Mockable> Mockable for DapiClientError<TE> {
     }
 }
 
-#[async_trait]
-/// DAPI client executor trait.
-pub trait DapiRequestExecutor {
-    /// Execute request using this DAPI client.
-    async fn execute<R>(
-        &self,
-        request: R,
-        settings: RequestSettings,
-    ) -> Result<R::Response, DapiClientError<<R::Client as TransportClient>::Error>>
-    where
-        R: TransportRequest + Mockable,
-        R::Response: Mockable,
-        <R::Client as TransportClient>::Error: Mockable;
-}
-
 /// Access point to DAPI.
 #[derive(Debug, Clone)]
 pub struct DapiClient {
@@ -126,7 +110,7 @@ impl DapiRequestExecutor for DapiClient {
         &self,
         request: R,
         settings: RequestSettings,
-    ) -> Result<R::Response, DapiClientError<<R::Client as TransportClient>::Error>>
+    ) -> ExecutionResult<R::Response, DapiClientError<<R::Client as TransportClient>::Error>>
     where
         R: TransportRequest + Mockable,
         R::Response: Mockable,
@@ -140,24 +124,25 @@ impl DapiRequestExecutor for DapiClient {
             .finalize();
 
         // Setup retry policy:
-        let retry_settings = ExponentialBuilder::default()
+        let retry_settings = ConstantBuilder::default()
             .with_max_times(applied_settings.retries)
-            // backon doesn't accept 1.0
-            .with_factor(1.001)
-            .with_min_delay(Duration::from_secs(0))
-            .with_max_delay(Duration::from_secs(0));
+            .with_delay(Duration::from_millis(10));
 
         // Save dump dir for later use, as self is moved into routine
         #[cfg(feature = "dump")]
         let dump_dir = self.dump_dir.clone();
         #[cfg(feature = "dump")]
         let dump_request = request.clone();
 
+        let retries_counter_arc = Arc::new(AtomicUsize::new(0));
+        let retries_counter_arc_ref = &retries_counter_arc;
+
         // Setup DAPI request execution routine future. It's a closure that will be called
         // more once to build new future on each retry.
         let routine = move || {
-            // Try to get an address to initialize transport on:
+            let retries_counter = Arc::clone(retries_counter_arc_ref);
 
+            // Try to get an address to initialize transport on:
             let address_list = self
                 .address_list
                 .read()
@@ -192,30 +177,29 @@ impl DapiRequestExecutor for DapiClient {
             async move {
                 // It stays wrapped in `Result` since we want to return
                 // `impl Future<Output = Result<...>`, not a `Result` itself.
-                let address = address_result?;
+                let address = address_result.map_err(|inner| ExecutionError {
+                    inner,
+                    retries: retries_counter.load(std::sync::atomic::Ordering::Acquire),
+                    address: None,
+                })?;
+
                 let pool = self.pool.clone();
 
                 let mut transport_client = R::Client::with_uri_and_settings(
                     address.uri().clone(),
                     &applied_settings,
                     &pool,
                 )
-                .map_err(|e| {
-                    DapiClientError::<<R::Client as TransportClient>::Error>::Transport(
-                        e,
-                        address.clone(),
-                    )
+                .map_err(|error| ExecutionError {
+                    inner: DapiClientError::Transport(error),
+                    retries: retries_counter.load(std::sync::atomic::Ordering::Acquire),
+                    address: Some(address.clone()),
                 })?;
 
                 let response = transport_request
                     .execute_transport(&mut transport_client, &applied_settings)
                     .await
-                    .map_err(|e| {
-                        DapiClientError::<<R::Client as TransportClient>::Error>::Transport(
-                            e,
-                            address.clone(),
-                        )
-                    });
+                    .map_err(DapiClientError::Transport);
 
                 match &response {
                     Ok(_) => {
@@ -226,8 +210,14 @@ impl DapiRequestExecutor for DapiClient {
                                 .write()
                                 .expect("can't get address list for write");
 
-                            address_list.unban_address(&address)
-                                .map_err(DapiClientError::<<R::Client as TransportClient>::Error>::AddressList)?;
+                            address_list.unban_address(&address).map_err(|error| {
+                                ExecutionError {
+                                    inner: DapiClientError::AddressList(error),
+                                    retries: retries_counter
+                                        .load(std::sync::atomic::Ordering::Acquire),
+                                    address: Some(address.clone()),
+                                }
+                            })?;
                         }
 
                         tracing::trace!(?response, "received {} response", response_name);
@@ -240,16 +230,34 @@ impl DapiRequestExecutor for DapiClient {
                                     .write()
                                     .expect("can't get address list for write");
 
-                                address_list.ban_address(&address)
-                                    .map_err(DapiClientError::<<R::Client as TransportClient>::Error>::AddressList)?;
+                                address_list.ban_address(&address).map_err(|error| {
+                                    ExecutionError {
+                                        inner: DapiClientError::AddressList(error),
+                                        retries: retries_counter
+                                            .load(std::sync::atomic::Ordering::Acquire),
+                                        address: Some(address.clone()),
+                                    }
+                                })?;
                             }
                         } else {
                             tracing::trace!(?error, "received error");
                         }
                     }
                 };
 
+                let retries = retries_counter.load(std::sync::atomic::Ordering::Acquire);
+
                 response
+                    .map(|inner| ExecutionResponse {
+                        inner,
+                        retries,
+                        address: address.clone(),
+                    })
+                    .map_err(|inner| ExecutionError {
+                        inner,
+                        retries,
+                        address: Some(address),
+                    })
             }
         };
 
@@ -258,11 +266,14 @@ impl DapiRequestExecutor for DapiClient {
         let result = routine
             .retry(retry_settings)
             .notify(|error, duration| {
+                let retries_counter = Arc::clone(&retries_counter_arc);
+                retries_counter.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+
                 tracing::warn!(
                     ?error,
                     "retrying error with sleeping {} secs",
                     duration.as_secs_f32()
-                )
+                );
             })
             .when(|e| e.can_retry())
             .instrument(tracing::info_span!("request routine"))

packages/rs-dapi-client/src/executor.rs

@@ -0,0 +1,66 @@
+use crate::transport::{TransportClient, TransportRequest};
+use crate::{Address, CanRetry, DapiClientError, RequestSettings};
+use dapi_grpc::mock::Mockable;
+use dapi_grpc::tonic::async_trait;
+use std::fmt::Debug;
+
+#[async_trait]
+/// DAPI client executor trait.
+pub trait DapiRequestExecutor {
+    /// Execute request using this DAPI client.
+    async fn execute<R>(
+        &self,
+        request: R,
+        settings: RequestSettings,
+    ) -> ExecutionResult<R::Response, DapiClientError<<R::Client as TransportClient>::Error>>
+    where
+        R: TransportRequest + Mockable,
+        R::Response: Mockable,
+        <R::Client as TransportClient>::Error: Mockable;
+}
+
+/// Error happened during request execution.
+#[derive(Debug, Clone, thiserror::Error, Eq, PartialEq)]
+#[error("{inner}")]
+pub struct ExecutionError<E> {
+    /// The cause of error
+    pub inner: E,
+    /// How many times the request was retried
+    pub retries: usize,
+    /// The address of the node that was used for the request
+    pub address: Option<Address>,
+}
+
+impl<E> ExecutionError<E> {
+    /// Unwrap the error cause
+    pub fn into_inner(self) -> E {
+        self.inner
+    }
+}
+
+impl<E: CanRetry> CanRetry for ExecutionError<E> {
+    fn can_retry(&self) -> bool {
+        self.inner.can_retry()
+    }
+}
+
+/// Request execution response.
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub struct ExecutionResponse<R> {
+    /// The response from the request
+    pub inner: R,
+    /// How many times the request was retried
+    pub retries: usize,
+    /// The address of the node that was used for the request
+    pub address: Address,
+}
+
+impl<R> ExecutionResponse<R> {
+    /// Unwrap the response
+    pub fn into_inner(self) -> R {
+        self.inner
+    }
+}
+
+/// Result of request execution
+pub type ExecutionResult<R, E> = Result<ExecutionResponse<R>, ExecutionError<E>>;

packages/rs-dapi-client/src/lib.rs

@@ -7,6 +7,7 @@ mod connection_pool;
 mod dapi_client;
 #[cfg(feature = "dump")]
 pub mod dump;
+mod executor;
 #[cfg(feature = "mocks")]
 pub mod mock;
 mod request_settings;
@@ -15,11 +16,11 @@ pub mod transport;
 pub use address_list::Address;
 pub use address_list::AddressList;
 pub use connection_pool::ConnectionPool;
-pub use dapi_client::DapiRequestExecutor;
 pub use dapi_client::{DapiClient, DapiClientError};
 use dapi_grpc::mock::Mockable;
 #[cfg(feature = "dump")]
 pub use dump::DumpData;
+pub use executor::*;
 use futures::{future::BoxFuture, FutureExt};
 pub use request_settings::RequestSettings;
 
@@ -48,7 +49,7 @@ pub trait DapiRequest {
         self,
         dapi_client: &'c D,
         settings: RequestSettings,
-    ) -> BoxFuture<'c, Result<Self::Response, DapiClientError<Self::TransportError>>>
+    ) -> BoxFuture<'c, ExecutionResult<Self::Response, DapiClientError<Self::TransportError>>>
     where
         Self: 'c;
 }
@@ -63,7 +64,7 @@ impl<T: transport::TransportRequest + Send> DapiRequest for T {
         self,
         dapi_client: &'c D,
         settings: RequestSettings,
-    ) -> BoxFuture<'c, Result<Self::Response, DapiClientError<Self::TransportError>>>
+    ) -> BoxFuture<'c, ExecutionResult<Self::Response, DapiClientError<Self::TransportError>>>
     where
         Self: 'c,
     {