sui_rpc/client/
mod.rs

1use std::sync::Arc;
2use std::time::Duration;
3use tap::Pipe;
4use tonic::body::Body;
5use tonic::codec::CompressionEncoding;
6use tonic::transport::channel::ClientTlsConfig;
7use tower::Layer;
8use tower::Service;
9use tower::ServiceBuilder;
10use tower::util::BoxLayer;
11use tower::util::BoxService;
12
13mod response_ext;
14pub use response_ext::ResponseExt;
15
16mod interceptors;
17pub use interceptors::HeadersInterceptor;
18
19mod watchdog;
20pub use watchdog::BodyIdleTimeout;
21use watchdog::DEFAULT_BODY_IDLE_TIMEOUT;
22use watchdog::WatchdogLayer;
23
24mod staking_rewards;
25pub use staking_rewards::DelegatedStake;
26
27mod coin_selection;
28mod lists;
29
30mod transaction_execution;
31pub use transaction_execution::ExecuteAndWaitError;
32
33use crate::proto::sui::rpc::v2::ledger_service_client::LedgerServiceClient;
34use crate::proto::sui::rpc::v2::move_package_service_client::MovePackageServiceClient;
35use crate::proto::sui::rpc::v2::signature_verification_service_client::SignatureVerificationServiceClient;
36use crate::proto::sui::rpc::v2::state_service_client::StateServiceClient;
37use crate::proto::sui::rpc::v2::subscription_service_client::SubscriptionServiceClient;
38use crate::proto::sui::rpc::v2::transaction_execution_service_client::TransactionExecutionServiceClient;
39#[cfg(feature = "unstable")]
40use crate::proto::sui::rpc::v2alpha::proof_service_client::ProofServiceClient;
41
42type Result<T, E = tonic::Status> = std::result::Result<T, E>;
43type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
44type BoxedChannel = BoxService<http::Request<Body>, http::Response<Body>, tonic::Status>;
45
46type RequestLayer = BoxLayer<
47    BoxService<http::Request<Body>, http::Response<Body>, BoxError>,
48    http::Request<Body>,
49    http::Response<Body>,
50    BoxError,
51>;
52
53const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
54const DEFAULT_TCP_KEEPALIVE_IDLE: Duration = Duration::from_secs(15);
55const DEFAULT_TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(5);
56const DEFAULT_TCP_KEEPALIVE_RETRIES: u32 = 3;
57const DEFAULT_HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(5);
58const DEFAULT_HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(20);
59
60// All RPCs made through a `Client` (and all of its clones) are multiplexed
61// over a single HTTP/2 connection, so the connection-level receive window is
62// shared by every in-flight response. A streaming response that the
63// application holds without polling pins up to a full stream window of that
64// shared budget; once the connection window is exhausted, every RPC on the
65// channel hangs indefinitely while TCP and HTTP/2 keepalives stay healthy.
66// hyper's client defaults (2 MiB stream / 5 MiB connection) let ~3 stalled
67// streams starve the connection. Keep the stream window at hyper's default
68// but raise the connection window so ~32 concurrently stalled streams are
69// needed instead.
70const DEFAULT_HTTP2_STREAM_WINDOW_SIZE: u32 = 2 * 1024 * 1024;
71const DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE: u32 = 64 * 1024 * 1024;
72
73/// A gRPC client for the Sui fullnode RPC interface.
74///
75/// All RPCs made through a client and its clones are multiplexed over a
76/// single HTTP/2 connection.
77///
78/// # Timeouts and deadlines
79///
80/// No default bounds the total duration of a call. Two opt-in bounds are
81/// available:
82///
83/// - A per-call deadline set with [`tonic::Request::set_timeout`]. This
84///   attaches the standard `grpc-timeout` header, so a server that supports
85///   it enforces the deadline too, and the client enforces it locally end to
86///   end: tonic bounds the wait for response headers, and the client's
87///   watchdog (see [`with_body_idle_timeout`](Client::with_body_idle_timeout))
88///   bounds the response body against the same deadline.
89/// - A client-wide response-headers timeout set with
90///   [`with_response_headers_timeout`](Client::with_response_headers_timeout).
91///   This is enforced locally only and its timer stops once response headers
92///   arrive, so it bounds every unary call (whose headers are not sent until
93///   the handler completes) without cutting off long-lived streams.
94///
95/// Independent of any deadline, the watchdog resets RPCs whose response body
96/// makes no progress for 30 seconds (configurable), so a call on a stalled
97/// connection fails with `DeadlineExceeded` instead of hanging forever.
98#[derive(Clone)]
99pub struct Client {
100    channel: tonic::transport::Channel,
101
102    // Everything other than the channel is only consulted when building a
103    // per-service client or reconfiguring, so it lives behind an `Arc` to
104    // keep `Client` itself small; it is cloned by value into futures
105    // throughout the SDK. The `Endpoint` alone is over 500 bytes.
106    config: Arc<ClientConfig>,
107}
108
109#[derive(Clone)]
110struct ClientConfig {
111    uri: http::Uri,
112    endpoint: tonic::transport::Endpoint,
113    headers: HeadersInterceptor,
114    max_decoding_message_size: Option<usize>,
115    body_idle_timeout: Option<Duration>,
116
117    /// Layer to apply to all RPC requests
118    request_layer: Option<RequestLayer>,
119}
120
121impl Client {
122    /// URL for the public-good, Sui Foundation provided fullnodes for mainnet.
123    pub const MAINNET_FULLNODE: &str = "https://fullnode.mainnet.sui.io";
124
125    /// URL for the public-good, Sui Foundation provided fullnodes for testnet.
126    pub const TESTNET_FULLNODE: &str = "https://fullnode.testnet.sui.io";
127
128    /// URL for the public-good, Sui Foundation provided fullnodes for devnet.
129    pub const DEVNET_FULLNODE: &str = "https://fullnode.devnet.sui.io";
130
131    /// URL for the public-good, Sui Foundation provided archive for mainnet.
132    pub const MAINNET_ARCHIVE: &str = "https://archive.mainnet.sui.io";
133
134    /// URL for the public-good, Sui Foundation provided archive for testnet.
135    pub const TESTNET_ARCHIVE: &str = "https://archive.testnet.sui.io";
136
137    /// Build a client from a fully custom [`tonic::transport::Endpoint`].
138    ///
139    /// This bypasses every transport default that [`Client::new`] applies,
140    /// including the HTTP/2 flow-control windows that protect the shared
141    /// connection from starvation by stalled streaming responses. Prefer
142    /// [`Client::new`] plus the `with_*` configuration methods unless an
143    /// endpoint setting is needed that the client does not expose. The
144    /// idle-body watchdog (see [`Client::with_body_idle_timeout`]) is part of
145    /// the client rather than the endpoint and stays enabled.
146    ///
147    /// In particular, do not rely on
148    /// [`http2_adaptive_window`](tonic::transport::Endpoint::http2_adaptive_window)
149    /// as a substitute for large static windows: with adaptive windowing,
150    /// hyper starts the connection window at the 64 KiB HTTP/2 spec default
151    /// until bandwidth-delay probing ramps up, so a single stalled stream can
152    /// starve the whole connection.
153    pub fn from_endpoint(endpoint: &tonic::transport::Endpoint) -> Self {
154        let uri = endpoint.uri().clone();
155        let channel = endpoint.connect_lazy();
156        Self {
157            channel,
158            config: Arc::new(ClientConfig {
159                uri,
160                endpoint: endpoint.clone(),
161                headers: Default::default(),
162                max_decoding_message_size: None,
163                body_idle_timeout: Some(DEFAULT_BODY_IDLE_TIMEOUT),
164                request_layer: None,
165            }),
166        }
167    }
168
169    #[allow(clippy::result_large_err)]
170    pub fn new<T>(uri: T) -> Result<Self>
171    where
172        T: TryInto<http::Uri>,
173        T::Error: Into<BoxError>,
174    {
175        let uri = uri
176            .try_into()
177            .map_err(Into::into)
178            .map_err(status_from_error)?;
179        let mut endpoint = tonic::transport::Endpoint::from(uri.clone());
180        if uri.scheme() == Some(&http::uri::Scheme::HTTPS) {
181            endpoint = endpoint
182                .tls_config(ClientTlsConfig::new().with_enabled_roots())
183                .map_err(Into::into)
184                .map_err(status_from_error)?;
185        }
186
187        let endpoint = endpoint
188            .connect_timeout(DEFAULT_CONNECT_TIMEOUT)
189            .tcp_keepalive(Some(DEFAULT_TCP_KEEPALIVE_IDLE))
190            .tcp_keepalive_interval(Some(DEFAULT_TCP_KEEPALIVE_INTERVAL))
191            .tcp_keepalive_retries(Some(DEFAULT_TCP_KEEPALIVE_RETRIES))
192            .http2_keep_alive_interval(DEFAULT_HTTP2_KEEP_ALIVE_INTERVAL)
193            .keep_alive_timeout(DEFAULT_HTTP2_KEEP_ALIVE_TIMEOUT)
194            .initial_stream_window_size(DEFAULT_HTTP2_STREAM_WINDOW_SIZE)
195            .initial_connection_window_size(DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE);
196        let channel = endpoint.connect_lazy();
197
198        Ok(Self {
199            channel,
200            config: Arc::new(ClientConfig {
201                uri,
202                endpoint,
203                headers: Default::default(),
204                max_decoding_message_size: None,
205                body_idle_timeout: Some(DEFAULT_BODY_IDLE_TIMEOUT),
206                request_layer: None,
207            }),
208        })
209    }
210
211    /// Set the idle timeout for the client's response-body watchdog.
212    /// Defaults to 30 seconds.
213    ///
214    /// The watchdog bounds the time between response-body progress events: if
215    /// a whole idle period passes without a frame of the response being
216    /// delivered to the caller -- because the connection is starved or dead,
217    /// or because the caller has parked a streaming response without polling
218    /// it -- the watchdog resets the stream, releasing the HTTP/2
219    /// flow-control window it had pinned, and the call observes a
220    /// [`DeadlineExceeded`](tonic::Code::DeadlineExceeded) status on its next
221    /// poll. This is what turns "an RPC on a starved connection hangs
222    /// forever" into a bounded failure, and what keeps an abandoned stream
223    /// from starving the shared connection in the first place.
224    ///
225    /// Streams that are legitimately quiet for longer than the timeout (the
226    /// fullnode's checkpoint subscription is not: it emits watermarks every
227    /// few seconds) should raise or disable the watchdog for that call with a
228    /// [`BodyIdleTimeout`] request extension.
229    pub fn with_body_idle_timeout(mut self, timeout: Duration) -> Self {
230        Arc::make_mut(&mut self.config).body_idle_timeout = Some(timeout);
231        self
232    }
233
234    /// Disable the client's response-body watchdog (see
235    /// [`with_body_idle_timeout`](Self::with_body_idle_timeout)).
236    ///
237    /// Without it, an RPC whose response can no longer make progress hangs
238    /// indefinitely; only disable the watchdog when every call is bounded by
239    /// the caller. It can be re-enabled for individual requests with a
240    /// [`BodyIdleTimeout`] request extension.
241    pub fn without_body_idle_timeout(mut self) -> Self {
242        Arc::make_mut(&mut self.config).body_idle_timeout = None;
243        self
244    }
245
246    /// Set a timeout for the response-headers phase of every RPC made
247    /// through this client. Disabled by default.
248    ///
249    /// The timer covers a request from dispatch on the connection until
250    /// response headers arrive and is dropped once they do, so a client-wide
251    /// value does not cut off long-lived streaming responses. Because a
252    /// server does not send response headers for a unary call until the
253    /// handler completes, this effectively bounds the total duration of
254    /// unary calls; the body that follows is bounded by the idle-body
255    /// watchdog (see [`with_body_idle_timeout`](Self::with_body_idle_timeout))
256    /// and, when set, the per-call deadline. Connection establishment is
257    /// bounded separately by the connect timeout.
258    ///
259    /// This timeout is enforced locally only; it is not communicated to the
260    /// server. When a per-call deadline ([`tonic::Request::set_timeout`]) is
261    /// also set, the shorter of the two bounds the headers phase locally, so
262    /// a per-call deadline can tighten this bound but never extend it --
263    /// size the timeout for the slowest expected RPC. Expiry surfaces as
264    /// [`DeadlineExceeded`](tonic::Code::DeadlineExceeded).
265    ///
266    /// This rebuilds the underlying channel, so it must be called before the
267    /// client is used or cloned; earlier clones keep the previous
268    /// configuration.
269    pub fn with_response_headers_timeout(mut self, timeout: Duration) -> Self {
270        let config = Arc::make_mut(&mut self.config);
271        config.endpoint = config.endpoint.clone().timeout(timeout);
272        self.channel = config.endpoint.connect_lazy();
273        self
274    }
275
276    /// Set the HTTP/2 per-stream receive window, in bytes.
277    ///
278    /// This bounds how much unread response data a single RPC can buffer
279    /// before the server must stop sending on that stream. It also bounds how
280    /// much of the shared connection window (see
281    /// [`with_initial_connection_window_size`](Self::with_initial_connection_window_size))
282    /// one stalled stream can pin. Defaults to 2 MiB.
283    ///
284    /// This rebuilds the underlying channel, so it must be called before the
285    /// client is used or cloned; earlier clones keep the previous
286    /// configuration.
287    pub fn with_initial_stream_window_size(mut self, size: u32) -> Self {
288        let config = Arc::make_mut(&mut self.config);
289        config.endpoint = config.endpoint.clone().initial_stream_window_size(size);
290        self.channel = config.endpoint.connect_lazy();
291        self
292    }
293
294    /// Set the HTTP/2 connection-level receive window, in bytes.
295    ///
296    /// This window is shared by every RPC multiplexed over the client's
297    /// single HTTP/2 connection, including all clones of the client. Response
298    /// data that the application has not yet read counts against it, so it
299    /// determines how many concurrently stalled streaming responses it takes
300    /// to starve the connection and hang every other RPC on it. Defaults to
301    /// 64 MiB (~32 stalled streams at the default 2 MiB stream window).
302    ///
303    /// This rebuilds the underlying channel, so it must be called before the
304    /// client is used or cloned; earlier clones keep the previous
305    /// configuration.
306    pub fn with_initial_connection_window_size(mut self, size: u32) -> Self {
307        let config = Arc::make_mut(&mut self.config);
308        config.endpoint = config.endpoint.clone().initial_connection_window_size(size);
309        self.channel = config.endpoint.connect_lazy();
310        self
311    }
312
313    pub fn with_headers(mut self, headers: HeadersInterceptor) -> Self {
314        Arc::make_mut(&mut self.config).headers = headers;
315        self
316    }
317
318    /// Provide an optional [`Layer`] that will be used to wrap all RPC
319    /// requests.
320    ///
321    /// This could be helpful in providing global metrics and logging
322    /// for all outbound requests.
323    ///
324    /// The layer's service may return any response body that implements
325    /// [`http_body::Body<Data = bytes::Bytes>`] and any error type that
326    /// implements `Into<Box<dyn Error + Send + Sync>>`. Both are mapped
327    /// to the internal types automatically.
328    ///
329    /// # Example
330    ///
331    /// Add a layer that logs each request URI:
332    ///
333    /// ```
334    /// # let _rt = tokio::runtime::Builder::new_current_thread()
335    /// #     .build()
336    /// #     .unwrap();
337    /// # let _guard = _rt.enter();
338    /// use sui_rpc::Client;
339    /// use tower::ServiceBuilder;
340    ///
341    /// let client = Client::new(Client::MAINNET_FULLNODE)
342    ///     .unwrap()
343    ///     .request_layer(ServiceBuilder::new().map_request(|req: http::Request<_>| {
344    ///         println!("request to {}", req.uri());
345    ///         req
346    ///     }));
347    /// ```
348    pub fn request_layer<L, ResBody, E>(mut self, layer: L) -> Self
349    where
350        L: Layer<BoxService<http::Request<Body>, http::Response<Body>, BoxError>>
351            + Send
352            + Sync
353            + 'static,
354        L::Service: Service<http::Request<Body>, Response = http::Response<ResBody>, Error = E>
355            + Send
356            + 'static,
357        <L::Service as Service<http::Request<Body>>>::Future: Send + 'static,
358        ResBody: http_body::Body<Data = bytes::Bytes> + Send + 'static,
359        ResBody::Error: Into<BoxError>,
360        E: Into<BoxError> + Send + 'static,
361    {
362        let layer = BoxLayer::new(
363            ServiceBuilder::new()
364                .map_response(|resp: http::Response<ResBody>| resp.map(Body::new))
365                .map_err(Into::<BoxError>::into)
366                .layer(layer),
367        );
368        Arc::make_mut(&mut self.config).request_layer = Some(layer);
369        self
370    }
371
372    pub fn with_max_decoding_message_size(mut self, limit: usize) -> Self {
373        Arc::make_mut(&mut self.config).max_decoding_message_size = Some(limit);
374        self
375    }
376
377    pub fn uri(&self) -> &http::Uri {
378        &self.config.uri
379    }
380
381    fn channel(&self) -> BoxedChannel {
382        let headers = self.config.headers.clone();
383
384        // Build the base service with headers applied at the HTTP level and the
385        // transport error mapped to BoxError for compatibility with user layers.
386        let base = BoxService::new(
387            ServiceBuilder::new()
388                .map_err(|e: tonic::transport::Error| -> BoxError { Box::new(e) })
389                .map_request(move |mut req: http::Request<Body>| {
390                    if !headers.headers().is_empty() {
391                        req.headers_mut()
392                            .extend(headers.headers().clone().into_headers());
393                    }
394                    req
395                })
396                .service(self.channel.clone()),
397        );
398
399        // Guard every response body with the idle-body watchdog, beneath any
400        // user layers so their view of the response goes through the
401        // watchdog's bridge.
402        let base = BoxService::new(WatchdogLayer::new(self.config.body_idle_timeout).layer(base));
403
404        // Apply the user's outbound request layer if present.
405        let layered = if let Some(layer) = &self.config.request_layer {
406            layer.layer(base)
407        } else {
408            base
409        };
410
411        // Map the final error to tonic::Status (a concrete type) so that
412        // downstream users of the tonic-generated clients don't run into
413        // lifetime-inference issues with async_trait and Box<dyn Error>.
414        BoxService::new(
415            ServiceBuilder::new()
416                .map_err(status_from_error)
417                .service(layered),
418        )
419    }
420
421    pub fn ledger_client(&mut self) -> LedgerServiceClient<BoxedChannel> {
422        LedgerServiceClient::new(self.channel())
423            .accept_compressed(CompressionEncoding::Zstd)
424            .pipe(|client| {
425                if let Some(limit) = self.config.max_decoding_message_size {
426                    client.max_decoding_message_size(limit)
427                } else {
428                    client
429                }
430            })
431    }
432
433    pub fn state_client(&mut self) -> StateServiceClient<BoxedChannel> {
434        StateServiceClient::new(self.channel())
435            .accept_compressed(CompressionEncoding::Zstd)
436            .pipe(|client| {
437                if let Some(limit) = self.config.max_decoding_message_size {
438                    client.max_decoding_message_size(limit)
439                } else {
440                    client
441                }
442            })
443    }
444
445    pub fn execution_client(&mut self) -> TransactionExecutionServiceClient<BoxedChannel> {
446        TransactionExecutionServiceClient::new(self.channel())
447            .accept_compressed(CompressionEncoding::Zstd)
448            .pipe(|client| {
449                if let Some(limit) = self.config.max_decoding_message_size {
450                    client.max_decoding_message_size(limit)
451                } else {
452                    client
453                }
454            })
455    }
456
457    pub fn package_client(&mut self) -> MovePackageServiceClient<BoxedChannel> {
458        MovePackageServiceClient::new(self.channel())
459            .accept_compressed(CompressionEncoding::Zstd)
460            .pipe(|client| {
461                if let Some(limit) = self.config.max_decoding_message_size {
462                    client.max_decoding_message_size(limit)
463                } else {
464                    client
465                }
466            })
467    }
468
469    pub fn signature_verification_client(
470        &mut self,
471    ) -> SignatureVerificationServiceClient<BoxedChannel> {
472        SignatureVerificationServiceClient::new(self.channel())
473            .accept_compressed(CompressionEncoding::Zstd)
474            .pipe(|client| {
475                if let Some(limit) = self.config.max_decoding_message_size {
476                    client.max_decoding_message_size(limit)
477                } else {
478                    client
479                }
480            })
481    }
482
483    pub fn subscription_client(&mut self) -> SubscriptionServiceClient<BoxedChannel> {
484        SubscriptionServiceClient::new(self.channel())
485            .accept_compressed(CompressionEncoding::Zstd)
486            .pipe(|client| {
487                if let Some(limit) = self.config.max_decoding_message_size {
488                    client.max_decoding_message_size(limit)
489                } else {
490                    client
491                }
492            })
493    }
494
495    /// Returns a client for the unstable alpha `ProofService`, which serves
496    /// Object Checkpoint State (OCS) inclusion proofs.
497    #[cfg(feature = "unstable")]
498    #[cfg_attr(doc_cfg, doc(cfg(feature = "unstable")))]
499    pub fn proof_client(&mut self) -> ProofServiceClient<BoxedChannel> {
500        ProofServiceClient::new(self.channel())
501            .accept_compressed(CompressionEncoding::Zstd)
502            .pipe(|client| {
503                if let Some(limit) = self.config.max_decoding_message_size {
504                    client.max_decoding_message_size(limit)
505                } else {
506                    client
507                }
508            })
509    }
510}
511
512/// Map a transport error to a [`tonic::Status`].
513///
514/// tonic surfaces an expired headers-phase timeout (the client's
515/// response-headers timeout, or a `grpc-timeout` deadline expiring before
516/// response headers arrive) as `Cancelled`. The gRPC code for an expired
517/// deadline is `DeadlineExceeded`, and the watchdog already uses it for
518/// body-phase expiry, so normalize before delegating to tonic's own mapping.
519fn status_from_error(error: BoxError) -> tonic::Status {
520    let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error.as_ref());
521    while let Some(err) = source {
522        // An embedded `Status` takes precedence, as in tonic's own mapping.
523        if err.is::<tonic::Status>() {
524            break;
525        }
526        if err.is::<tonic::TimeoutExpired>() {
527            return tonic::Status::deadline_exceeded(
528                "timeout expired before response headers were received",
529            );
530        }
531        source = err.source();
532    }
533    tonic::Status::from_error(error)
534}
sui_rpc/client/mod.rs

sui_rpc/client/
mod.rs
