feat(sdk): sdk-level retry logic for fetch and fetch_many

packages/rs-dapi-client/Cargo.toml

@@ -5,7 +5,6 @@ edition = "2021"
 
 [features]
 default = ["mocks", "offline-testing"]
-tokio-sleep = ["backon/tokio-sleep"]
 mocks = [
     "dep:sha2",
     "dep:hex",
@@ -20,7 +19,7 @@ dump = ["mocks"]
 offline-testing = []
 
 [dependencies]
-backon = { version = "1.2"}
+backon = { version = "1.2", features = ["tokio-sleep"] }
 dapi-grpc = { path = "../dapi-grpc", features = [
     "core",
     "platform",

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"))