Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 37 commits into from
Oct 28, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
37 commits
Select commit Hold shift + click to select a range
7a77ccf
feat(sdk): provide request execution information
shumkov Oct 21, 2024
b85f52e
test: fix compilation errors
shumkov Oct 21, 2024
6de23ab
refactor: rename `unwrap` to `into_inner`
shumkov Oct 21, 2024
fba6ef8
refactor: rename response to inner
shumkov Oct 21, 2024
8af6776
test: increase timeout to pass the test
shumkov Oct 21, 2024
8e278e3
fix: `into_inner` doesn't need `CanRetry` bound
shumkov Oct 21, 2024
5af8822
fix: errored address is none
shumkov Oct 21, 2024
8269c1f
Merge branch 'v1.4-dev' into feat/sdk/execution_info
shumkov Oct 22, 2024
a9a96a0
chore: don't use empty uri
shumkov Oct 22, 2024
920b50d
docs: fix usage example
shumkov Oct 22, 2024
067db87
refactor: remove unused import
shumkov Oct 22, 2024
c684b3f
chore: update test vectors
shumkov Oct 22, 2024
9da89a8
chore: more test vectors
shumkov Oct 22, 2024
34ffdfb
feat(sdk): retry failed requests on sdk level
lklimek Oct 21, 2024
60d220c
fix(dapi-client): impl VersionedGrpcResponse for ExecutionResponse
lklimek Oct 22, 2024
80d19e0
feat(sdk): fech with retries
lklimek Oct 23, 2024
f78555e
chore(sdk): remove unnecessary changes
lklimek Oct 24, 2024
4343b6e
fix(sdk): correct retry logic
lklimek Oct 24, 2024
50ad84c
feat(sdk): retry impl
lklimek Oct 24, 2024
e4e667a
chore: self-review
lklimek Oct 24, 2024
ac388a8
chore: apply feedback
lklimek Oct 25, 2024
016450f
chore: remove outdated comment
lklimek Oct 25, 2024
0738f76
chore(sdk): impl<T> From<ExecutionError<T>> for Error
lklimek Oct 25, 2024
438185f
doc(sdk): comment on sync::retry() logic
lklimek Oct 25, 2024
170d5a9
chore: self review
lklimek Oct 25, 2024
e20a29d
refactor: remove unused code
lklimek Oct 25, 2024
c9c6c2f
chore: fmt
lklimek Oct 25, 2024
e5e17a7
chore: doc
lklimek Oct 25, 2024
dace7fa
test: fix retry test
lklimek Oct 25, 2024
eca69f8
refactor(sdk): apply feedback
lklimek Oct 25, 2024
df706f4
test(sdk): add edge case to test_retry
lklimek Oct 25, 2024
1f954d5
fix: use settings in fetch_many
lklimek Oct 25, 2024
034bdb2
test(sdk): regenerate test vectors
lklimek Oct 28, 2024
1bc31a9
refactor: re-export specific types
shumkov Oct 28, 2024
5a497ca
Merge remote-tracking branch 'origin/feat/sdk/execution_info' into fe…
lklimek Oct 28, 2024
723b1c5
chore: fixes after merge
lklimek Oct 28, 2024
cebc709
Merge remote-tracking branch 'origin/v1.4-dev' into feat/sdk-retry
lklimek Oct 28, 2024
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 1 addition & 2 deletions packages/rs-dapi-client/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ edition = "2021"

[features]
default = ["mocks", "offline-testing"]
tokio-sleep = ["backon/tokio-sleep"]
mocks = [
"dep:sha2",
"dep:hex",
Expand All @@ -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",
Expand Down
94 changes: 88 additions & 6 deletions packages/rs-dapi-client/src/executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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}")]
Expand All @@ -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()
}
}

Expand All @@ -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()),
}
}
}
4 changes: 3 additions & 1 deletion packages/rs-dapi-client/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
25 changes: 25 additions & 0 deletions packages/rs-dapi-client/src/mock.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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()
}
}
}
3 changes: 2 additions & 1 deletion packages/rs-dapi-client/src/request_settings.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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>,
Expand Down
2 changes: 1 addition & 1 deletion packages/rs-sdk/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down
9 changes: 3 additions & 6 deletions packages/rs-sdk/src/core/transaction.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};

Expand Down Expand Up @@ -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()))
}

Expand Down Expand Up @@ -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()?;
shumkov marked this conversation as resolved.
Show resolved Hide resolved

core_chain_locked_height = height;

Expand Down
13 changes: 12 additions & 1 deletion packages/rs-sdk/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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())
}
}
shumkov marked this conversation as resolved.
Show resolved Hide resolved

impl CanRetry for Error {
fn can_retry(&self) -> bool {
matches!(self, Error::StaleNode(..) | Error::TimeoutReached(_, _))
Expand Down
77 changes: 42 additions & 35 deletions packages/rs-sdk/src/platform/fetch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
})
};
shumkov marked this conversation as resolved.
Show resolved Hide resolved

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()
shumkov marked this conversation as resolved.
Show resolved Hide resolved
}

/// Fetch single object from Platform.
Expand Down
Loading
Loading