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    /// The lowest protocol version at which every node of a network serves this request as
75    /// asked. Below it some nodes run a release that does not know the request's method or
76    /// one of its fields: they answer UNIMPLEMENTED, or drop the field and answer a different
77    /// question, and either gets them banned. `None` when nodes of every version serve it.
78    ///
79    /// The SDK's fetch paths refuse such a request until the network has proven that
80    /// version; a caller that executes the request directly owns the check.
81    fn needs_protocol_version(&self) -> Option<u32> {
82        None
83    }
84
85    /// Perform transport request asynchronously.
86    fn execute_transport<'c>(
87        self,
88        client: &'c mut Self::Client,
89        settings: &AppliedRequestSettings,
90    ) -> BoxFuture<'c, Result<Self::Response, TransportError>>;
91}
92
93/// Transport error type.
94#[derive(Debug, thiserror::Error)]
95#[cfg_attr(feature = "mocks", derive(serde::Serialize, serde::Deserialize))]
96pub enum TransportError {
97    /// gRPC error
98    #[error("grpc error: {0}")]
99    Grpc(
100        #[from]
101        #[cfg_attr(feature = "mocks", serde(with = "dapi_grpc::mock::serde_mockable"))]
102        dapi_grpc::tonic::Status,
103    ),
104}
105
106impl Clone for TransportError {
107    fn clone(&self) -> Self {
108        match self {
109            TransportError::Grpc(status) => {
110                // tonic::Status doesn't implement Clone, so we reconstruct it
111                // from its components. Note: this loses the original error source.
112                let cloned_status = dapi_grpc::tonic::Status::with_details_and_metadata(
113                    status.code(),
114                    status.message(),
115                    status.details().to_vec().into(),
116                    status.metadata().clone(),
117                );
118                TransportError::Grpc(cloned_status)
119            }
120        }
121    }
122}
123
124impl CanRetry for TransportError {
125    fn can_retry(&self) -> bool {
126        match self {
127            TransportError::Grpc(status) => status.can_retry(),
128        }
129    }
130
131    fn rate_limit_ban_duration(&self) -> Option<std::time::Duration> {
132        match self {
133            TransportError::Grpc(status) => status.rate_limit_ban_duration(),
134        }
135    }
136}
137
138/// Serialization of [TransportError].
139///
140/// We need to do manual serialization because of the generic type parameter which doesn't support serde derive.
141impl Mockable for TransportError {
142    #[cfg(feature = "mocks")]
143    fn mock_serialize(&self) -> Option<Vec<u8>> {
144        Some(serde_json::to_vec(self).expect("serialize Transport error"))
145    }
146
147    #[cfg(feature = "mocks")]
148    fn mock_deserialize(data: &[u8]) -> Option<Self> {
149        Some(serde_json::from_slice(data).expect("deserialize Transport error"))
150    }
151}
152
153/// Serialization of boxed [TransportError].
154impl Mockable for Box<TransportError> {
155    #[cfg(feature = "mocks")]
156    fn mock_serialize(&self) -> Option<Vec<u8>> {
157        self.as_ref().mock_serialize()
158    }
159
160    #[cfg(feature = "mocks")]
161    fn mock_deserialize(data: &[u8]) -> Option<Self> {
162        TransportError::mock_deserialize(data).map(Box::new)
163    }
164}
165
166/// Generic way to create a transport client from provided [Uri].
167pub trait TransportClient: Send + Sized {
168    /// Build client using node's url.
169    fn with_uri(uri: Uri, pool: &ConnectionPool) -> Result<Self, TransportError>;
170
171    /// Build client using node's url and [AppliedRequestSettings].
172    fn with_uri_and_settings(
173        uri: Uri,
174        settings: &AppliedRequestSettings,
175        pool: &ConnectionPool,
176    ) -> Result<Self, TransportError>;
177
178    /// Build client using node's url and [AppliedRequestSettings], together
179    /// with the pool generation of the connection it took (see
180    /// [ConnectionPool::generation]). When an attempt misses its deadline,
181    /// the executor evicts the node's connections up to that generation.
182    ///
183    /// The default reads the pool's generation after building the client, so
184    /// it can also cover a connection another thread pooled in between.
185    /// Clients that take their connection from `pool` override it with the
186    /// generation [ConnectionPool::get_or_create_with_generation] returns.
187    fn with_uri_and_settings_and_generation(
188        uri: Uri,
189        settings: &AppliedRequestSettings,
190        pool: &ConnectionPool,
191    ) -> Result<(Self, u64), TransportError> {
192        let client = Self::with_uri_and_settings(uri, settings, pool)?;
193        Ok((client, pool.generation()))
194    }
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200    use dapi_grpc::tonic::Code;
201
202    #[test]
203    fn test_tonic_status_can_retry_retryable_codes() {
204        let retryable_codes = vec![
205            Code::Ok,
206            Code::DataLoss,
207            Code::Cancelled,
208            Code::Unknown,
209            Code::DeadlineExceeded,
210            Code::ResourceExhausted,
211            Code::Aborted,
212            Code::Internal,
213            Code::Unavailable,
214            // Retryable on another node: an old build in a mixed-version network
215            // returns Unimplemented for methods newer nodes already serve.
216            Code::Unimplemented,
217        ];
218
219        for code in retryable_codes {
220            let status = dapi_grpc::tonic::Status::new(code, "test");
221            assert!(
222                status.can_retry(),
223                "Expected code {:?} to be retryable",
224                code
225            );
226        }
227    }
228
229    #[test]
230    fn test_tonic_status_can_retry_non_retryable_codes() {
231        let non_retryable_codes = vec![
232            Code::InvalidArgument,
233            Code::NotFound,
234            Code::AlreadyExists,
235            Code::PermissionDenied,
236            Code::FailedPrecondition,
237            Code::OutOfRange,
238            Code::Unauthenticated,
239        ];
240
241        for code in non_retryable_codes {
242            let status = dapi_grpc::tonic::Status::new(code, "test");
243            assert!(
244                !status.can_retry(),
245                "Expected code {:?} to be non-retryable",
246                code
247            );
248        }
249    }
250
251    #[test]
252    fn test_transport_error_can_retry() {
253        let retryable = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("temporary"));
254        assert!(retryable.can_retry());
255
256        let non_retryable = TransportError::Grpc(dapi_grpc::tonic::Status::not_found("permanent"));
257        assert!(!non_retryable.can_retry());
258    }
259
260    /// `rate_limit_ban_duration` returns `Some` only for `ResourceExhausted` with
261    /// a parseable positive `ratelimit-reset` header.  Every other code returns
262    /// `None` regardless of headers.
263    #[test]
264    fn test_tonic_status_rate_limit_ban_duration() {
265        use dapi_grpc::tonic::metadata::MetadataValue;
266
267        // ResourceExhausted + valid header → Some.
268        let mut status = dapi_grpc::tonic::Status::new(Code::ResourceExhausted, "429");
269        status
270            .metadata_mut()
271            .insert("ratelimit-reset", MetadataValue::try_from("30").unwrap());
272        assert_eq!(
273            status.rate_limit_ban_duration(),
274            Some(std::time::Duration::from_secs(30))
275        );
276
277        // ResourceExhausted without header → None.
278        let no_header = dapi_grpc::tonic::Status::new(Code::ResourceExhausted, "429");
279        assert!(no_header.rate_limit_ban_duration().is_none());
280
281        // Non-ResourceExhausted codes → None regardless.
282        for code in [
283            Code::Ok,
284            Code::Unavailable,
285            Code::Internal,
286            Code::DeadlineExceeded,
287        ] {
288            let mut s = dapi_grpc::tonic::Status::new(code, "x");
289            s.metadata_mut()
290                .insert("ratelimit-reset", MetadataValue::try_from("30").unwrap());
291            assert!(
292                s.rate_limit_ban_duration().is_none(),
293                "code {code:?} must return None"
294            );
295        }
296    }
297
298    #[test]
299    fn test_transport_error_rate_limit_ban_duration_delegates() {
300        use dapi_grpc::tonic::metadata::MetadataValue;
301
302        let mut status = dapi_grpc::tonic::Status::new(Code::ResourceExhausted, "429");
303        status
304            .metadata_mut()
305            .insert("ratelimit-reset", MetadataValue::try_from("45").unwrap());
306        let rate_limited = TransportError::Grpc(status);
307        assert_eq!(
308            rate_limited.rate_limit_ban_duration(),
309            Some(std::time::Duration::from_secs(45))
310        );
311        // Still retryable — rate-limit ban duration doesn't affect can_retry.
312        assert!(rate_limited.can_retry());
313
314        let unavailable = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("down"));
315        assert!(unavailable.rate_limit_ban_duration().is_none());
316        assert!(unavailable.can_retry());
317    }
318
319    #[test]
320    fn test_transport_error_clone() {
321        let original = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("test message"));
322
323        let cloned = original.clone();
324
325        match (&original, &cloned) {
326            (TransportError::Grpc(orig), TransportError::Grpc(clone)) => {
327                assert_eq!(orig.code(), clone.code());
328                assert_eq!(orig.message(), clone.message());
329            }
330        }
331    }
332
333    #[test]
334    fn test_transport_error_display() {
335        let err = TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("service down"));
336        let display = format!("{}", err);
337        assert!(display.contains("service down"));
338    }
339
340    #[cfg(feature = "mocks")]
341    #[test]
342    fn test_transport_error_mock_roundtrip() {
343        let original =
344            TransportError::Grpc(dapi_grpc::tonic::Status::unavailable("test roundtrip"));
345        let serialized = original.mock_serialize().expect("should serialize");
346        let deserialized =
347            TransportError::mock_deserialize(&serialized).expect("should deserialize");
348
349        match deserialized {
350            TransportError::Grpc(status) => {
351                assert_eq!(status.code(), Code::Unavailable);
352            }
353        }
354    }
355
356    #[cfg(feature = "mocks")]
357    #[test]
358    fn test_boxed_transport_error_mock_roundtrip() {
359        let original = Box::new(TransportError::Grpc(dapi_grpc::tonic::Status::internal(
360            "boxed test",
361        )));
362        let serialized = original.mock_serialize().expect("should serialize");
363        let deserialized =
364            Box::<TransportError>::mock_deserialize(&serialized).expect("should deserialize");
365
366        match *deserialized {
367            TransportError::Grpc(status) => {
368                assert_eq!(status.code(), Code::Internal);
369            }
370        }
371    }
372}
rs_dapi_client/transport.rs

rs_dapi_client/
transport.rs
