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/executor.rs

@@ -19,6 +19,21 @@ pub trait DapiRequestExecutor {
         <R::Client as TransportClient>::Error: Mockable;
 }
 
+/// Unwrap wrapped types
+pub trait IntoInner<T> {
+    /// Unwrap the inner type.
+    ///
+    /// This function returns inner type, dropping additional context information.
+    /// It is lossy operation, so it should be used with caution.
+    fn into_inner(self) -> T;
+}
+
+/// Convert inner type without loosing additional context information of the wrapper.
+pub trait InnerInto<T> {
+    /// Convert inner type without loosing additional context information of the wrapper.
+    fn inner_into(self) -> T;
+}
+
 /// Error happened during request execution.
 #[derive(Debug, Clone, thiserror::Error, Eq, PartialEq)]
 #[error("{inner}")]
@@ -31,10 +46,27 @@ pub struct ExecutionError<E> {
     pub address: Option<Address>,
 }
 
-impl<E> ExecutionError<E> {
+impl<F, T> InnerInto<ExecutionError<T>> for ExecutionError<F>
+where
+    F: Into<T>,
+{
+    /// Convert inner error type without loosing retries and address
+    fn inner_into(self) -> ExecutionError<T> {
+        ExecutionError {
+            inner: self.inner.into(),
+            retries: self.retries,
+            address: self.address,
+        }
+    }
+}
+
+impl<E, I> IntoInner<I> for ExecutionError<E>
+where
+    E: Into<I>,
+{
     /// Unwrap the error cause
-    pub fn into_inner(self) -> E {
-        self.inner
+    fn into_inner(self) -> I {
+        self.inner.into()
     }
 }
 
@@ -55,12 +87,62 @@ pub struct ExecutionResponse<R> {
     pub address: Address,
 }
 
-impl<R> ExecutionResponse<R> {
+#[cfg(feature = "mocks")]
+impl<R: Default> Default for ExecutionResponse<R> {
+    fn default() -> Self {
+        Self {
+            retries: Default::default(),
+            address: "http://127.0.0.1".parse().expect("create mock address"),
+            inner: Default::default(),
+        }
+    }
+}
+
+impl<R, I> IntoInner<I> for ExecutionResponse<R>
+where
+    R: Into<I>,
+{
     /// Unwrap the response
-    pub fn into_inner(self) -> R {
-        self.inner
+    fn into_inner(self) -> I {
+        self.inner.into()
+    }
+}
+
+impl<F, T> InnerInto<ExecutionResponse<T>> for ExecutionResponse<F>
+where
+    F: Into<T>,
+{
+    /// Convert inner response type without loosing retries and address
+    fn inner_into(self) -> ExecutionResponse<T> {
+        ExecutionResponse {
+            inner: self.inner.into(),
+            retries: self.retries,
+            address: self.address,
+        }
     }
 }
 
 /// Result of request execution
 pub type ExecutionResult<R, E> = Result<ExecutionResponse<R>, ExecutionError<E>>;
+
+impl<R, E> IntoInner<Result<R, E>> for ExecutionResult<R, E> {
+    fn into_inner(self) -> Result<R, E> {
+        match self {
+            Ok(response) => Ok(response.into_inner()),
+            Err(error) => Err(error.into_inner()),
+        }
+    }
+}
+
+impl<F, FE, T, TE> InnerInto<ExecutionResult<T, TE>> for ExecutionResult<F, FE>
+where
+    F: Into<T>,
+    FE: Into<TE>,
+{
+    fn inner_into(self) -> ExecutionResult<T, TE> {
+        match self {
+            Ok(response) => Ok(response.inner_into()),
+            Err(error) => Err(error.inner_into()),
+        }
+    }
+}

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

@@ -21,7 +21,9 @@ pub use dapi_client::{DapiClient, DapiClientError};
 use dapi_grpc::mock::Mockable;
 #[cfg(feature = "dump")]
 pub use dump::DumpData;
-pub use executor::{DapiRequestExecutor, ExecutionError, ExecutionResponse, ExecutionResult};
+pub use executor::{
+    DapiRequestExecutor, ExecutionError, ExecutionResponse, ExecutionResult, InnerInto, IntoInner,
+};
 use futures::{future::BoxFuture, FutureExt};
 pub use request_settings::RequestSettings;
 

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

@@ -290,3 +290,28 @@ impl<E: Mockable> Mockable for ExecutionError<E> {
         })
     }
 }
+
+/// Create full wrapping object from inner type, using defaults for
+/// fields that cannot be derived from the inner type.
+pub trait FromInner<R>
+where
+    Self: Default,
+{
+    /// Create full wrapping object from inner type, using defaults for
+    /// fields that cannot be derived from the inner type.
+    ///
+    /// Note this is imprecise conversion and should be avoided outside of tests.
+    fn from_inner(inner: R) -> Self;
+}
+
+impl<R> FromInner<R> for ExecutionResponse<R>
+where
+    Self: Default,
+{
+    fn from_inner(inner: R) -> Self {
+        Self {
+            inner,
+            ..Default::default()
+        }
+    }
+}

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

@@ -21,7 +21,8 @@ pub struct RequestSettings {
     pub connect_timeout: Option<Duration>,
     /// Timeout for a request.
     pub timeout: Option<Duration>,
-    /// Number of retries until returning the last error.
+    /// Number of retries in case of failed requests. If max retries reached, the last error is returned.
+    /// 1 means one request and one retry in case of error, etc.
     pub retries: Option<usize>,
     /// Ban DAPI address if node not responded or responded with error.
     pub ban_failed_address: Option<bool>,

packages/rs-sdk/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 
 arc-swap = { version = "1.7.1" }
+backon = { version = "1.2", features = ["tokio-sleep"] }
 chrono = { version = "0.4.38" }
 dpp = { path = "../rs-dpp", default-features = false, features = [
   "dash-sdk-features",
@@ -57,7 +58,6 @@ test-case = { version = "3.3.1" }
 
 [features]
 default = ["mocks", "offline-testing"]
-tokio-sleep = ["rs-dapi-client/tokio-sleep"]
 
 mocks = [
   "dep:serde",

packages/rs-sdk/src/core/transaction.rs

@@ -12,7 +12,7 @@ use dpp::identity::state_transition::asset_lock_proof::chain::ChainAssetLockProo
 use dpp::identity::state_transition::asset_lock_proof::InstantAssetLockProof;
 use dpp::prelude::AssetLockProof;
 
-use rs_dapi_client::{DapiRequestExecutor, RequestSettings};
+use rs_dapi_client::{DapiRequestExecutor, IntoInner, RequestSettings};
 use std::time::Duration;
 use tokio::time::{sleep, timeout};
 
@@ -56,9 +56,7 @@ impl Sdk {
         };
         self.execute(core_transactions_stream, RequestSettings::default())
             .await
-            // TODO: We need better way to handle execution response and errors
-            .map(|execution_response| execution_response.into_inner())
-            .map_err(|execution_error| execution_error.into_inner())
+            .into_inner()
             .map_err(|e| Error::DapiClientError(e.to_string()))
     }
 
@@ -184,8 +182,7 @@ impl Sdk {
                                     RequestSettings::default(),
                                 )
                                 .await // TODO: We need better way to handle execution errors
-                                .map_err(|error| error.into_inner())?
-                                .into_inner();
+                                .into_inner()?;
 
                             core_chain_locked_height = height;
 

packages/rs-sdk/src/error.rs

@@ -5,11 +5,12 @@ use std::time::Duration;
 use dapi_grpc::mock::Mockable;
 use dpp::version::PlatformVersionError;
 use dpp::ProtocolError;
-use rs_dapi_client::{CanRetry, DapiClientError};
+use rs_dapi_client::{CanRetry, DapiClientError, ExecutionError};
 
 pub use drive_proof_verifier::error::ContextProviderError;
 
 /// Error type for the SDK
+// TODO: Propagate server address and retry information so that the user can retrieve it
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     /// SDK is not configured properly
@@ -85,6 +86,16 @@ impl From<PlatformVersionError> for Error {
     }
 }
 
+impl<T> From<ExecutionError<T>> for Error
+where
+    ExecutionError<T>: ToString,
+{
+    fn from(value: ExecutionError<T>) -> Self {
+        // TODO: Improve error handling
+        Self::DapiClientError(value.to_string())
+    }
+}
+
 impl CanRetry for Error {
     fn can_retry(&self) -> bool {
         matches!(self, Error::StaleNode(..) | Error::TimeoutReached(_, _))

packages/rs-sdk/src/platform/fetch.rs

@@ -9,6 +9,7 @@
 //!   traits. The associated [Fetch::Request]` type needs to implement [TransportRequest].
 
 use crate::mock::MockResponse;
+use crate::sync::retry;
 use crate::{error::Error, platform::query::Query, Sdk};
 use dapi_grpc::platform::v0::{self as platform_proto, Proof, ResponseMetadata};
 use dpp::voting::votes::Vote;
@@ -18,6 +19,7 @@ use dpp::{
 };
 use drive_proof_verifier::FromProof;
 use rs_dapi_client::{transport::TransportRequest, DapiRequest, RequestSettings};
+use rs_dapi_client::{ExecutionError, ExecutionResponse, InnerInto, IntoInner};
 use std::fmt::Debug;
 
 use super::types::identity::IdentityRequest;
@@ -119,25 +121,9 @@ where
         query: Q,
         settings: Option<RequestSettings>,
     ) -> Result<(Option<Self>, ResponseMetadata), Error> {
-        let request = query.query(sdk.prove())?;
-
-        let response = request
-            .clone()
-            .execute(sdk, settings.unwrap_or_default())
-            .await // TODO: We need better way to handle execution response and errors
-            .map(|execution_response| execution_response.into_inner())
-            .map_err(|execution_error| execution_error.into_inner())?;
-
-        let object_type = std::any::type_name::<Self>().to_string();
-        tracing::trace!(request = ?request, response = ?response, object_type, "fetched object from platform");
-
-        let (object, response_metadata): (Option<Self>, ResponseMetadata) =
-            sdk.parse_proof_with_metadata(request, response).await?;
-
-        match object {
-            Some(item) => Ok((item.into(), response_metadata)),
-            None => Ok((None, response_metadata)),
-        }
+        Self::fetch_with_metadata_and_proof(sdk, query, settings)
+            .await
+            .map(|(object, metadata, _)| (object, metadata))
     }
 
     /// Fetch single object from Platform with metadata and underlying proof.
@@ -169,26 +155,47 @@ where
         query: Q,
         settings: Option<RequestSettings>,
     ) -> Result<(Option<Self>, ResponseMetadata, Proof), Error> {
-        let request = query.query(sdk.prove())?;
+        let request: &<Self as Fetch>::Request = &query.query(sdk.prove())?;
+
+        let fut = |settings: RequestSettings| async move {
+            let ExecutionResponse {
+                address,
+                retries,
+                inner: response,
+            } = request
+                .clone()
+                .execute(sdk, settings)
+                .await
+                .map_err(|execution_error| execution_error.inner_into())?;
+
+            let object_type = std::any::type_name::<Self>().to_string();
+            tracing::trace!(request = ?request, response = ?response, ?address, retries, object_type, "fetched object from platform");
 
-        let response = request
-            .clone()
-            .execute(sdk, settings.unwrap_or_default())
-            .await // TODO: We need better way to handle execution response and errors
-            .map(|execution_response| execution_response.into_inner())
-            .map_err(|execution_error| execution_error.into_inner())?;
+            let (object, response_metadata, proof): (Option<Self>, ResponseMetadata, Proof) = sdk
+                .parse_proof_with_metadata_and_proof(request.clone(), response)
+                .await
+                .map_err(|e| ExecutionError {
+                    inner: e,
+                    address: Some(address.clone()),
+                    retries,
+                })?;
 
-        let object_type = std::any::type_name::<Self>().to_string();
-        tracing::trace!(request = ?request, response = ?response, object_type, "fetched object from platform");
+            match object {
+                Some(item) => Ok((item.into(), response_metadata, proof)),
+                None => Ok((None, response_metadata, proof)),
+            }
+            .map(|x| ExecutionResponse {
+                inner: x,
+                address,
+                retries,
+            })
+        };
 
-        let (object, response_metadata, proof): (Option<Self>, ResponseMetadata, Proof) = sdk
-            .parse_proof_with_metadata_and_proof(request, response)
-            .await?;
+        let settings = sdk
+            .dapi_client_settings
+            .override_by(settings.unwrap_or_default());
 
-        match object {
-            Some(item) => Ok((item.into(), response_metadata, proof)),
-            None => Ok((None, response_metadata, proof)),
-        }
+        retry(settings, fut).await.into_inner()
     }
 
     /// Fetch single object from Platform.