rs_dapi_client/
transport.rs

1//! Transport options that DAPI requests use under the hood.
2
3pub(crate) mod grpc;
4#[cfg(not(target_arch = "wasm32"))]
5pub(crate) mod tonic_channel;
6#[cfg(target_arch = "wasm32")]
7pub(crate) mod wasm_channel;
8
9use crate::connection_pool::ConnectionPool;
10pub use crate::request_settings::AppliedRequestSettings;
11use crate::{CanRetry, RequestSettings, Uri};
12use dapi_grpc::mock::Mockable;
13pub use futures::future::BoxFuture;
14use std::any;
15use std::fmt::Debug;
16use std::time::Duration;
17
18#[cfg(not(target_arch = "wasm32"))]
19pub use tonic_channel::{
20    create_channel, CoreGrpcClient, PlatformGrpcClient, TokioBackonSleeper as BackonSleeper,
21};
22#[cfg(target_arch = "wasm32")]
23pub use wasm_channel::{
24    create_channel, CoreGrpcClient, PlatformGrpcClient, WasmBackonSleeper as BackonSleeper,
25};
26
27/// Sleep for the given duration.
28#[cfg(not(target_arch = "wasm32"))]
29pub async fn sleep(duration: Duration) {
30    tokio::time::sleep(duration).await;
31}
32
33/// Sleep for the given duration.
34#[cfg(target_arch = "wasm32")]
35pub async fn sleep(duration: Duration) {
36    wasm_channel::into_send_sleep(duration).await;
37}
38
39/// Generic transport layer request.
40/// Requires [Clone] as could be retried and a client in general consumes a request.
41pub trait TransportRequest: Clone + Send + Sync + Debug + Mockable {
42    /// A client specific to this type of transport.
43    type Client: TransportClient;
44
45    /// Transport layer response.
46    type Response: Mockable + Send + Debug;
47
48    /// Settings that will override [DapiClient](crate::DapiClient)'s ones each time the request is executed.
49    const SETTINGS_OVERRIDES: RequestSettings;
50
51    /// gRPC request name
52    fn request_name(&self) -> &'static str {
53        any::type_name::<Self>()
54    }
55
56    /// gRPC response name
57    fn response_name(&self) -> &'static str {
58        any::type_name::<Self::Response>()
59    }
60
61    /// gRPC method name
62    fn method_name(&self) -> &'static str;
63
64    /// When the request asks a proof of nothing (an empty id list), the INVALID_ARGUMENT
65    /// message every node refuses it with. [DapiClient](crate::DapiClient) then refuses it
66    /// the same way without sending it: a node released before that refusal answers with an
67    /// internal error, or with a proof of nothing the client cannot verify, and either one
68    /// gets the node banned and the request sent to the next. Without a proof, nodes answer
69    /// an empty list with an empty result, so the request is sent.
70    fn names_nothing(&self) -> Option<&'static str> {
71        None
72    }
73
74    /// Perform transport request asynchronously.
75    fn execute_transport<'c>(
76        self,
77        client: &'c mut Self::Client,
78        settings: &AppliedRequestSettings,
79    ) -> BoxFuture<'c, Result<Self::Response, TransportError>>;
80}
81
82/// Transport error type.
83#[derive(Debug, thiserror::Error)]
84#[cfg_attr(feature = "mocks", derive(serde::Serialize, serde::Deserialize))]
85pub enum TransportError {
86    /// gRPC error
87    #[error("grpc error: {0}")]
88    Grpc(
89        #[from]
90        #[cfg_attr(feature = "mocks", serde(with = "dapi_grpc::mock::serde_mockable"))]
91        dapi_grpc::tonic::Status,
92    ),
93}
94
95impl Clone for TransportError {
96    fn clone(&self) -> Self {
97        match self {
98            TransportError::Grpc(status) => {
99                // tonic::Status doesn't implement Clone, so we reconstruct it
100                // from its components. Note: this loses the original error source.
101                let cloned_status = dapi_grpc::tonic::Status::with_details_and_metadata(
102                    status.code(),
103                    status.message(),
104                    status.details().to_vec().into(),
105                    status.metadata().clone(),
106                );
107                TransportError::Grpc(cloned_status)
108            }
109        }
110    }
111}
112
113impl CanRetry for TransportError {
114    fn can_retry(&self) -> bool {
115        match self {
116            TransportError::Grpc(status) => status.can_retry(),
117        }
118    }
119
120    fn rate_limit_ban_duration(&self) -> Option<std::time::Duration> {
121        match self {
122            TransportError::Grpc(status) => status.rate_limit_ban_duration(),
123        }
124    }
125}
126
127/// Serialization of [TransportError].
128///
129/// We need to do manual serialization because of the generic type parameter which doesn't support serde derive.
130impl Mockable for TransportError {
131    #[cfg(feature = "mocks")]
132    fn mock_serialize(&self) -> Option<Vec<u8>> {
133        Some(serde_json::to_vec(self).expect("serialize Transport error"))
134    }
135
136    #[cfg(feature = "mocks")]
137    fn mock_deserialize(data: &[u8]) -> Option<Self> {
138        Some(serde_json::from_slice(data).expect("deserialize Transport error"))
139    }
140}
141
142/// Serialization of boxed [TransportError].
143impl Mockable for Box<TransportError> {
144    #[cfg(feature = "mocks")]
145    fn mock_serialize(&self) -> Option<Vec<u8>> {
146        self.as_ref().mock_serialize()
147    }
148
149    #[cfg(feature = "mocks")]
150    fn mock_deserialize(data: &[u8]) -> Option<Self> {
151        TransportError::mock_deserialize(data).map(Box::new)
152    }
153}
154
155/// Generic way to create a transport client from provided [Uri].
156pub trait TransportClient: Send + Sized {
157    /// Build client using node's url.
158    fn with_uri(uri: Uri, pool: &ConnectionPool) -> Result<Self, TransportError>;
159
160    /// Build client using node's url and [AppliedRequestSettings].
161    fn with_uri_and_settings(
162        uri: Uri,
163        settings: &AppliedRequestSettings,
164        pool: &ConnectionPool,
165    ) -> Result<Self, TransportError>;
166
167    /// Build client using node's url and [AppliedRequestSettings], together
168    /// with the pool generation of the connection it took (see
169    /// [ConnectionPool::generation]). When an attempt misses its deadline,
170    /// the executor evicts the node's connections up to that generation.
171    ///
172    /// The default reads the pool's generation after building the client, so
173    /// it can also cover a connection another thread pooled in between.
174    /// Clients that take their connection from `pool` override it with the
175    /// generation [ConnectionPool::get_or_create_with_generation] returns.
176    fn with_uri_and_settings_and_generation(
177        uri: Uri,
178        settings: &AppliedRequestSettings,
179        pool: &ConnectionPool,
180    ) -> Result<(Self, u64), TransportError> {
181        let client = Self::with_uri_and_settings(uri, settings, pool)?;
182        Ok((client, pool.generation()))
183    }
184}
185
186#[cfg(test)]
187mod tests {
188    use super::*;
189    use dapi_grpc::tonic::Code;
190
191    #[test]
192    fn test_tonic_status_can_retry_retryable_codes() {
193        let retryable_codes = vec![
194            Code::Ok,
195            Code::DataLoss,
196            Code::Cancelled,
197            Code::Unknown,
198            Code::DeadlineExceeded,
199            Code::ResourceExhausted,
200            Code::Aborted,
201            Code::Internal,
202            Code::Unavailable,
203            // Retryable on another node: an old build in a mixed-version network
204            // returns Unimplemented for methods newer nodes already serve.
205            Code::Unimplemented,
206        ];
207
208        for code in retryable_codes {
209            let status = dapi_grpc::tonic::Status::new(code, "test");
210            assert!(
211                status.can_retry(),
212                "Expected code {:?} to be retryable",
213                code
214            );
215        }
216    }
217
218    #[test]
219    fn test_tonic_status_can_retry_non_retryable_codes() {
220        let non_retryable_codes = vec![
221            Code::InvalidArgument,
222            Code::NotFound,
223            Code::AlreadyExists,
224            Code::PermissionDenied,
225            Code::FailedPrecondition,
226            Code::OutOfRange,
227            Code::Unauthenticated,
228        ];
229
230        for code in non_retryable_codes {
231            let status = dapi_grpc::tonic::Status::new(code, "test");
232            assert!(
233                !status.can_retry(),
234                "Expected code {:?} to be non-retryable",
235                code
236            );
237        }
238    }
239
240    #[test]
241    fn test_transport_error_can_retry() {
242        let retryable = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("temporary"));
243        assert!(retryable.can_retry());
244
245        let non_retryable = TransportError::Grpc(dapi_grpc::tonic::Status::not_found("permanent"));
246        assert!(!non_retryable.can_retry());
247    }
248
249    /// `rate_limit_ban_duration` returns `Some` only for `ResourceExhausted` with
250    /// a parseable positive `ratelimit-reset` header.  Every other code returns
251    /// `None` regardless of headers.
252    #[test]
253    fn test_tonic_status_rate_limit_ban_duration() {
254        use dapi_grpc::tonic::metadata::MetadataValue;
255
256        // ResourceExhausted + valid header → Some.
257        let mut status = dapi_grpc::tonic::Status::new(Code::ResourceExhausted, "429");
258        status
259            .metadata_mut()
260            .insert("ratelimit-reset", MetadataValue::try_from("30").unwrap());
261        assert_eq!(
262            status.rate_limit_ban_duration(),
263            Some(std::time::Duration::from_secs(30))
264        );
265
266        // ResourceExhausted without header → None.
267        let no_header = dapi_grpc::tonic::Status::new(Code::ResourceExhausted, "429");
268        assert!(no_header.rate_limit_ban_duration().is_none());
269
270        // Non-ResourceExhausted codes → None regardless.
271        for code in [
272            Code::Ok,
273            Code::Unavailable,
274            Code::Internal,
275            Code::DeadlineExceeded,
276        ] {
277            let mut s = dapi_grpc::tonic::Status::new(code, "x");
278            s.metadata_mut()
279                .insert("ratelimit-reset", MetadataValue::try_from("30").unwrap());
280            assert!(
281                s.rate_limit_ban_duration().is_none(),
282                "code {code:?} must return None"
283            );
284        }
285    }
286
287    #[test]
288    fn test_transport_error_rate_limit_ban_duration_delegates() {
289        use dapi_grpc::tonic::metadata::MetadataValue;
290
291        let mut status = dapi_grpc::tonic::Status::new(Code::ResourceExhausted, "429");
292        status
293            .metadata_mut()
294            .insert("ratelimit-reset", MetadataValue::try_from("45").unwrap());
295        let rate_limited = TransportError::Grpc(status);
296        assert_eq!(
297            rate_limited.rate_limit_ban_duration(),
298            Some(std::time::Duration::from_secs(45))
299        );
300        // Still retryable — rate-limit ban duration doesn't affect can_retry.
301        assert!(rate_limited.can_retry());
302
303        let unavailable = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("down"));
304        assert!(unavailable.rate_limit_ban_duration().is_none());
305        assert!(unavailable.can_retry());
306    }
307
308    #[test]
309    fn test_transport_error_clone() {
310        let original = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("test message"));
311
312        let cloned = original.clone();
313
314        match (&original, &cloned) {
315            (TransportError::Grpc(orig), TransportError::Grpc(clone)) => {
316                assert_eq!(orig.code(), clone.code());
317                assert_eq!(orig.message(), clone.message());
318            }
319        }
320    }
321
322    #[test]
323    fn test_transport_error_display() {
324        let err = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("service down"));
325        let display = format!("{}", err);
326        assert!(display.contains("service down"));
327    }
328
329    #[cfg(feature = "mocks")]
330    #[test]
331    fn test_transport_error_mock_roundtrip() {
332        let original =
333            TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("test roundtrip"));
334        let serialized = original.mock_serialize().expect("should serialize");
335        let deserialized =
336            TransportError::mock_deserialize(&serialized).expect("should deserialize");
337
338        match deserialized {
339            TransportError::Grpc(status) => {
340                assert_eq!(status.code(), Code::Unavailable);
341            }
342        }
343    }
344
345    #[cfg(feature = "mocks")]
346    #[test]
347    fn test_boxed_transport_error_mock_roundtrip() {
348        let original = Box::new(TransportError::Grpc(dapi_grpc::tonic::Status::internal(
349            "boxed test",
350        )));
351        let serialized = original.mock_serialize().expect("should serialize");
352        let deserialized =
353            Box::<TransportError>::mock_deserialize(&serialized).expect("should deserialize");
354
355        match *deserialized {
356            TransportError::Grpc(status) => {
357                assert_eq!(status.code(), Code::Internal);
358            }
359        }
360    }
361}
rs_dapi_client/transport.rs

rs_dapi_client/
transport.rs
