Skip to main content

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 ledger_streams;
29mod lists;
30pub use ledger_streams::CheckpointStreamFrame;
31pub use ledger_streams::CheckpointStreamRequest;
32pub use ledger_streams::CheckpointStreamStart;
33pub use ledger_streams::Delivery;
34pub use ledger_streams::EventStreamFrame;
35pub use ledger_streams::EventStreamRequest;
36pub use ledger_streams::EventStreamStart;
37pub use ledger_streams::LedgerStreamConfig;
38pub use ledger_streams::LedgerStreamEvent;
39pub use ledger_streams::LedgerStreamFamily;
40pub use ledger_streams::LedgerStreamOperation;
41pub use ledger_streams::LedgerStreamStage;
42pub use ledger_streams::ListConfig;
43pub use ledger_streams::ListEvent;
44pub use ledger_streams::TransactionStreamFrame;
45pub use ledger_streams::TransactionStreamRequest;
46pub use ledger_streams::TransactionStreamStart;
47
48mod transaction_execution;
49pub use transaction_execution::ExecuteAndWaitError;
50
51use crate::proto::sui::rpc::v2::ledger_service_client::LedgerServiceClient;
52use crate::proto::sui::rpc::v2::move_package_service_client::MovePackageServiceClient;
53use crate::proto::sui::rpc::v2::signature_verification_service_client::SignatureVerificationServiceClient;
54use crate::proto::sui::rpc::v2::state_service_client::StateServiceClient;
55use crate::proto::sui::rpc::v2::subscription_service_client::SubscriptionServiceClient;
56use crate::proto::sui::rpc::v2::transaction_execution_service_client::TransactionExecutionServiceClient;
57#[cfg(feature = "unstable")]
58use crate::proto::sui::rpc::v2alpha::proof_service_client::ProofServiceClient;
59
60type Result<T, E = tonic::Status> = std::result::Result<T, E>;
61type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
62type BoxedChannel = BoxService<http::Request<Body>, http::Response<Body>, tonic::Status>;
63
64type RequestLayer = BoxLayer<
65    BoxService<http::Request<Body>, http::Response<Body>, BoxError>,
66    http::Request<Body>,
67    http::Response<Body>,
68    BoxError,
69>;
70
71const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
72const DEFAULT_TCP_KEEPALIVE_IDLE: Duration = Duration::from_secs(15);
73const DEFAULT_TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(5);
74const DEFAULT_TCP_KEEPALIVE_RETRIES: u32 = 3;
75const DEFAULT_HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(5);
76const DEFAULT_HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(20);
77const DEFAULT_NUM_CONNECTIONS: usize = 1;
78
79// RPCs made through a `Client` (and all of its clones) are multiplexed over
80// its HTTP/2 connections, so each connection-level receive window is shared
81// by every in-flight response on that connection. A streaming response that the
82// application holds without polling pins up to a full stream window of that
83// shared budget; once the connection window is exhausted, every RPC on the
84// channel hangs indefinitely while TCP and HTTP/2 keepalives stay healthy.
85// hyper's client defaults (2 MiB stream / 5 MiB connection) let ~3 stalled
86// streams starve the connection. Keep the stream window at hyper's default
87// but raise the connection window so ~32 concurrently stalled streams are
88// needed instead.
89const DEFAULT_HTTP2_STREAM_WINDOW_SIZE: u32 = 2 * 1024 * 1024;
90const DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE: u32 = 64 * 1024 * 1024;
91
92/// A gRPC client for the Sui fullnode RPC interface.
93///
94/// RPCs made through a client and its clones are multiplexed over a single
95/// HTTP/2 connection by default; see
96/// [`with_num_connections`](Client::with_num_connections) to spread them over
97/// several.
98///
99/// # Ledger streams
100///
101/// `Client` provides two styles of stream operations for checkpoints, transactions, and events:
102///
103/// - `list_*` ([`list_checkpoints`](Client::list_checkpoints), [`list_transactions`](Client::list_transactions), [`list_events`](Client::list_events)):
104///   Finite pagination streams yielding raw response pages until reaching an end bound or the ledger tip.
105/// - `stream_*` ([`stream_checkpoints`](Client::stream_checkpoints), [`stream_transactions`](Client::stream_transactions), [`stream_events`](Client::stream_events)):
106///   Resumable, infinite streams starting from any position (`Tip`, `Checkpoint`, or `Resume` cursor)
107///   that automatically handle backfill replay, live subscription or polling, and transient error retries.
108///
109/// Stream read masks must include `sequence_number` for checkpoints; `checkpoint` and
110/// `transaction_index` for transactions; and `checkpoint`, `transaction_index`, and `event_index`
111/// for events (`"*"` satisfies all three).
112///
113/// Process a payload before persisting its restart position. To resume after a checkpoint, pass
114/// `cursor + 1` to [`CheckpointStreamStart::Checkpoint`]. To resume transactions or events, pass
115/// `frame.cursor` to [`TransactionStreamStart::Resume`] or [`EventStreamStart::Resume`].
116///
117/// ## Example
118///
119/// ```no_run
120/// # use futures::StreamExt;
121/// # use sui_rpc::field::{FieldMask, FieldMaskUtil};
122/// # use sui_rpc::proto::sui::rpc::v2::ExecutedTransaction;
123/// # use sui_rpc::Client;
124/// # use sui_rpc::client::{TransactionStreamRequest, TransactionStreamStart};
125/// #
126/// # async fn process_transaction(_: &ExecutedTransaction) {}
127/// # async fn persist_position(_: &[u8]) {}
128/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
129/// let client = Client::new(Client::MAINNET_FULLNODE)?;
130/// let read_mask = FieldMask::from_paths(["checkpoint", "transaction_index", "digest"]);
131/// let request = TransactionStreamRequest::new().with_read_mask(read_mask.clone());
132///
133/// let mut stream = Box::pin(client.stream_transactions(request));
134/// let persisted = loop {
135///     let Some(frame) = stream.next().await else {
136///         return Ok(());
137///     };
138///     let frame = frame?;
139///
140///     if let Some(transaction) = &frame.transaction {
141///         process_transaction(transaction).await;
142///     }
143///     let persisted = frame.cursor.clone();
144///     persist_position(&persisted).await;
145///     break persisted;
146/// };
147/// drop(stream);
148///
149/// let restart_request = TransactionStreamRequest::new()
150///     .with_read_mask(read_mask)
151///     .with_start(TransactionStreamStart::Resume(persisted));
152/// let mut resumed = Box::pin(client.stream_transactions(restart_request));
153/// let _ = resumed.next().await;
154/// # Ok(())
155/// # }
156/// ```
157///
158/// # Timeouts and deadlines
159///
160/// No default bounds the total duration of a call. Two opt-in bounds are
161/// available:
162///
163/// - A per-call deadline set with [`tonic::Request::set_timeout`]. This
164///   attaches the standard `grpc-timeout` header, so a server that supports
165///   it enforces the deadline too, and the client enforces it locally end to
166///   end: tonic bounds the wait for response headers, and the client's
167///   watchdog (see [`with_body_idle_timeout`](Client::with_body_idle_timeout))
168///   bounds the response body against the same deadline.
169///   The ledger-stream `list_*` and `stream_*` methods do not accept
170///   `tonic::Request`. One logical operation may issue several RPCs, so per-RPC
171///   authentication, common headers, tracing, and timeouts belong on the
172///   client: static headers on [`Client::with_headers`], dynamic per-RPC auth
173///   or tracing on [`Client::request_layer`]. Configuring a [`Clone`]d client
174///   affects only that clone and shares the underlying connection, so a single
175///   stream gets dedicated headers by receiving a configured clone.
176/// - A client-wide response-headers timeout set with
177///   [`with_response_headers_timeout`](Client::with_response_headers_timeout).
178///   This is enforced locally only and its timer stops once response headers
179///   arrive, so it bounds every unary call (whose headers are not sent until
180///   the handler completes) without cutting off long-lived streams.
181///
182/// Independent of any deadline, the watchdog resets RPCs whose response body
183/// makes no progress for 30 seconds (configurable), so a call on a stalled
184/// connection fails with `DeadlineExceeded` instead of hanging forever.
185#[derive(Clone)]
186pub struct Client {
187    channel: tonic::transport::Channel,
188
189    // Everything other than the channel is only consulted when building a
190    // per-service client or reconfiguring, so it lives behind an `Arc` to
191    // keep `Client` itself small; it is cloned by value into futures
192    // throughout the SDK. The `Endpoint` alone is over 500 bytes.
193    config: Arc<ClientConfig>,
194}
195
196#[derive(Clone)]
197struct ClientConfig {
198    uri: http::Uri,
199    endpoint: tonic::transport::Endpoint,
200    headers: HeadersInterceptor,
201    max_decoding_message_size: Option<usize>,
202    body_idle_timeout: Option<Duration>,
203    num_connections: usize,
204
205    /// Layer to apply to all RPC requests
206    request_layer: Option<RequestLayer>,
207}
208
209/// Open `num_connections` lazy connections to `endpoint`, all sharing its
210/// configuration, and balance requests across them.
211///
212/// Each connection is inserted under its own key, because the balancer
213/// identifies backends by key and the endpoint is the same for all of them.
214/// The change sender is dropped once the set is populated: the set is fixed,
215/// and the balancer retains it after discovery ends.
216fn build_channel(
217    endpoint: &tonic::transport::Endpoint,
218    num_connections: usize,
219) -> tonic::transport::Channel {
220    if num_connections <= 1 {
221        return endpoint.connect_lazy();
222    }
223
224    let (channel, changes) = tonic::transport::Channel::balance_channel::<usize>(num_connections);
225    for key in 0..num_connections {
226        changes
227            .try_send(tonic::transport::channel::Change::Insert(
228                key,
229                endpoint.clone(),
230            ))
231            .expect("change channel has capacity for every connection");
232    }
233
234    channel
235}
236
237impl Client {
238    /// URL for the public-good, Sui Foundation provided fullnodes for mainnet.
239    pub const MAINNET_FULLNODE: &str = "https://fullnode.mainnet.sui.io";
240
241    /// URL for the public-good, Sui Foundation provided fullnodes for testnet.
242    pub const TESTNET_FULLNODE: &str = "https://fullnode.testnet.sui.io";
243
244    /// URL for the public-good, Sui Foundation provided fullnodes for devnet.
245    pub const DEVNET_FULLNODE: &str = "https://fullnode.devnet.sui.io";
246
247    /// URL for the public-good, Sui Foundation provided archive for mainnet.
248    pub const MAINNET_ARCHIVE: &str = "https://archive.mainnet.sui.io";
249
250    /// URL for the public-good, Sui Foundation provided archive for testnet.
251    pub const TESTNET_ARCHIVE: &str = "https://archive.testnet.sui.io";
252
253    /// Build a client from a fully custom [`tonic::transport::Endpoint`].
254    ///
255    /// This bypasses every transport default that [`Client::new`] applies,
256    /// including the HTTP/2 flow-control windows that protect a connection
257    /// from starvation by stalled streaming responses. Prefer
258    /// [`Client::new`] plus the `with_*` configuration methods unless an
259    /// endpoint setting is needed that the client does not expose. The
260    /// idle-body watchdog (see [`Client::with_body_idle_timeout`]) is part of
261    /// the client rather than the endpoint and stays enabled.
262    ///
263    /// In particular, do not rely on
264    /// [`http2_adaptive_window`](tonic::transport::Endpoint::http2_adaptive_window)
265    /// as a substitute for large static windows: with adaptive windowing,
266    /// hyper starts the connection window at the 64 KiB HTTP/2 spec default
267    /// until bandwidth-delay probing ramps up, so a single stalled stream can
268    /// starve the whole connection.
269    pub fn from_endpoint(endpoint: &tonic::transport::Endpoint) -> Self {
270        let uri = endpoint.uri().clone();
271        let channel = build_channel(endpoint, DEFAULT_NUM_CONNECTIONS);
272        Self {
273            channel,
274            config: Arc::new(ClientConfig {
275                uri,
276                endpoint: endpoint.clone(),
277                headers: Default::default(),
278                max_decoding_message_size: None,
279                body_idle_timeout: Some(DEFAULT_BODY_IDLE_TIMEOUT),
280                num_connections: DEFAULT_NUM_CONNECTIONS,
281                request_layer: None,
282            }),
283        }
284    }
285
286    #[allow(clippy::result_large_err)]
287    pub fn new<T>(uri: T) -> Result<Self>
288    where
289        T: TryInto<http::Uri>,
290        T::Error: Into<BoxError>,
291    {
292        let uri = uri
293            .try_into()
294            .map_err(Into::into)
295            .map_err(status_from_error)?;
296        let mut endpoint = tonic::transport::Endpoint::from(uri.clone());
297        if uri.scheme() == Some(&http::uri::Scheme::HTTPS) {
298            endpoint = endpoint
299                .tls_config(ClientTlsConfig::new().with_enabled_roots())
300                .map_err(Into::into)
301                .map_err(status_from_error)?;
302        }
303
304        let endpoint = endpoint
305            .connect_timeout(DEFAULT_CONNECT_TIMEOUT)
306            .tcp_keepalive(Some(DEFAULT_TCP_KEEPALIVE_IDLE))
307            .tcp_keepalive_interval(Some(DEFAULT_TCP_KEEPALIVE_INTERVAL))
308            .tcp_keepalive_retries(Some(DEFAULT_TCP_KEEPALIVE_RETRIES))
309            .http2_keep_alive_interval(DEFAULT_HTTP2_KEEP_ALIVE_INTERVAL)
310            .keep_alive_timeout(DEFAULT_HTTP2_KEEP_ALIVE_TIMEOUT)
311            .initial_stream_window_size(DEFAULT_HTTP2_STREAM_WINDOW_SIZE)
312            .initial_connection_window_size(DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE);
313        let channel = build_channel(&endpoint, DEFAULT_NUM_CONNECTIONS);
314
315        Ok(Self {
316            channel,
317            config: Arc::new(ClientConfig {
318                uri,
319                endpoint,
320                headers: Default::default(),
321                max_decoding_message_size: None,
322                body_idle_timeout: Some(DEFAULT_BODY_IDLE_TIMEOUT),
323                num_connections: DEFAULT_NUM_CONNECTIONS,
324                request_layer: None,
325            }),
326        })
327    }
328
329    /// Set the idle timeout for the client's response-body watchdog.
330    /// Defaults to 30 seconds.
331    ///
332    /// The watchdog bounds the time between response-body progress events: if
333    /// a whole idle period passes without a frame of the response being
334    /// delivered to the caller -- because the connection is starved or dead,
335    /// or because the caller has parked a streaming response without polling
336    /// it -- the watchdog resets the stream, releasing the HTTP/2
337    /// flow-control window it had pinned, and the call observes a
338    /// [`DeadlineExceeded`](tonic::Code::DeadlineExceeded) status on its next
339    /// poll. This is what turns "an RPC on a starved connection hangs
340    /// forever" into a bounded failure, and what keeps an abandoned stream
341    /// from starving the shared connection in the first place.
342    ///
343    /// Streams that are legitimately quiet for longer than the timeout (the
344    /// fullnode's checkpoint subscription is not: it emits watermarks every
345    /// few seconds) should raise or disable the watchdog for that call with a
346    /// [`BodyIdleTimeout`] request extension.
347    pub fn with_body_idle_timeout(mut self, timeout: Duration) -> Self {
348        Arc::make_mut(&mut self.config).body_idle_timeout = Some(timeout);
349        self
350    }
351
352    /// Disable the client's response-body watchdog (see
353    /// [`with_body_idle_timeout`](Self::with_body_idle_timeout)).
354    ///
355    /// Without it, an RPC whose response can no longer make progress hangs
356    /// indefinitely; only disable the watchdog when every call is bounded by
357    /// the caller. It can be re-enabled for individual requests with a
358    /// [`BodyIdleTimeout`] request extension.
359    pub fn without_body_idle_timeout(mut self) -> Self {
360        Arc::make_mut(&mut self.config).body_idle_timeout = None;
361        self
362    }
363
364    /// Set a timeout for the response-headers phase of every RPC made
365    /// through this client. Disabled by default.
366    ///
367    /// The timer covers a request from dispatch on the connection until
368    /// response headers arrive and is dropped once they do, so a client-wide
369    /// value does not cut off long-lived streaming responses. Because a
370    /// server does not send response headers for a unary call until the
371    /// handler completes, this effectively bounds the total duration of
372    /// unary calls; the body that follows is bounded by the idle-body
373    /// watchdog (see [`with_body_idle_timeout`](Self::with_body_idle_timeout))
374    /// and, when set, the per-call deadline. Connection establishment is
375    /// bounded separately by the connect timeout.
376    ///
377    /// This timeout is enforced locally only; it is not communicated to the
378    /// server. When a per-call deadline ([`tonic::Request::set_timeout`]) is
379    /// also set, the shorter of the two bounds the headers phase locally, so
380    /// a per-call deadline can tighten this bound but never extend it --
381    /// size the timeout for the slowest expected RPC. Expiry surfaces as
382    /// [`DeadlineExceeded`](tonic::Code::DeadlineExceeded).
383    ///
384    /// This rebuilds the underlying channel, so it must be called before the
385    /// client is used or cloned; earlier clones keep the previous
386    /// configuration.
387    pub fn with_response_headers_timeout(mut self, timeout: Duration) -> Self {
388        let config = Arc::make_mut(&mut self.config);
389        config.endpoint = config.endpoint.clone().timeout(timeout);
390        self.channel = build_channel(&config.endpoint, config.num_connections);
391        self
392    }
393
394    /// Set how many HTTP/2 connections the client opens to the endpoint.
395    /// Defaults to 1; a count of 0 is treated as 1.
396    ///
397    /// Requests are distributed across the connections rather than sharing
398    /// one connection's flow-control window and driver task. A single
399    /// connection's throughput is bounded by that window and by the one task
400    /// driving its multiplexed streams, so workloads that keep many streams
401    /// in flight at once (bulk or streaming reads) can be limited by it well
402    /// before the server is. Every connection carries the same endpoint
403    /// configuration, so the window sizes described in
404    /// [`with_initial_connection_window_size`](Self::with_initial_connection_window_size)
405    /// apply to each one, and the client's total window budget scales with
406    /// the count.
407    ///
408    /// Placement is effectively random per request, not a strict rotation, so
409    /// connections carry equal load only on average. Requests held open
410    /// concurrently, such as long-lived streams, can land unevenly.
411    ///
412    /// Connections are established lazily and reconnect independently, and
413    /// one that cannot be established fails only the requests routed to it.
414    ///
415    /// This rebuilds the underlying channel, so it must be called before the
416    /// client is used or cloned; earlier clones keep the previous
417    /// configuration.
418    pub fn with_num_connections(mut self, num_connections: usize) -> Self {
419        let config = Arc::make_mut(&mut self.config);
420        config.num_connections = num_connections.max(1);
421        self.channel = build_channel(&config.endpoint, config.num_connections);
422        self
423    }
424
425    /// Set the HTTP/2 per-stream receive window, in bytes.
426    ///
427    /// This bounds how much unread response data a single RPC can buffer
428    /// before the server must stop sending on that stream. It also bounds how
429    /// much of the shared connection window (see
430    /// [`with_initial_connection_window_size`](Self::with_initial_connection_window_size))
431    /// one stalled stream can pin. Defaults to 2 MiB.
432    ///
433    /// This rebuilds the underlying channel, so it must be called before the
434    /// client is used or cloned; earlier clones keep the previous
435    /// configuration.
436    pub fn with_initial_stream_window_size(mut self, size: u32) -> Self {
437        let config = Arc::make_mut(&mut self.config);
438        config.endpoint = config.endpoint.clone().initial_stream_window_size(size);
439        self.channel = build_channel(&config.endpoint, config.num_connections);
440        self
441    }
442
443    /// Set the HTTP/2 connection-level receive window, in bytes.
444    ///
445    /// This window is shared by every RPC multiplexed over one of the
446    /// client's HTTP/2 connections, including those issued by clones of the
447    /// client. Response data that the application has not yet read counts
448    /// against it, so it determines how many concurrently stalled streaming
449    /// responses it takes to starve a connection and hang every other RPC on
450    /// it. Defaults to 64 MiB (~32 stalled streams at the default 2 MiB
451    /// stream window). The window applies per connection, so a client
452    /// configured with
453    /// [`with_num_connections`](Self::with_num_connections) has this much on
454    /// each.
455    ///
456    /// This rebuilds the underlying channel, so it must be called before the
457    /// client is used or cloned; earlier clones keep the previous
458    /// configuration.
459    pub fn with_initial_connection_window_size(mut self, size: u32) -> Self {
460        let config = Arc::make_mut(&mut self.config);
461        config.endpoint = config.endpoint.clone().initial_connection_window_size(size);
462        self.channel = build_channel(&config.endpoint, config.num_connections);
463        self
464    }
465
466    pub fn with_headers(mut self, headers: HeadersInterceptor) -> Self {
467        Arc::make_mut(&mut self.config).headers = headers;
468        self
469    }
470
471    /// Provide an optional [`Layer`] that will be used to wrap all RPC
472    /// requests.
473    ///
474    /// This could be helpful in providing global metrics and logging
475    /// for all outbound requests.
476    ///
477    /// The layer's service may return any response body that implements
478    /// [`http_body::Body<Data = bytes::Bytes>`] and any error type that
479    /// implements `Into<Box<dyn Error + Send + Sync>>`. Both are mapped
480    /// to the internal types automatically.
481    ///
482    /// # Example
483    ///
484    /// Add a layer that logs each request URI:
485    ///
486    /// ```
487    /// # let _rt = tokio::runtime::Builder::new_current_thread()
488    /// #     .build()
489    /// #     .unwrap();
490    /// # let _guard = _rt.enter();
491    /// use sui_rpc::Client;
492    /// use tower::ServiceBuilder;
493    ///
494    /// let client = Client::new(Client::MAINNET_FULLNODE)
495    ///     .unwrap()
496    ///     .request_layer(ServiceBuilder::new().map_request(|req: http::Request<_>| {
497    ///         println!("request to {}", req.uri());
498    ///         req
499    ///     }));
500    /// ```
501    pub fn request_layer<L, ResBody, E>(mut self, layer: L) -> Self
502    where
503        L: Layer<BoxService<http::Request<Body>, http::Response<Body>, BoxError>>
504            + Send
505            + Sync
506            + 'static,
507        L::Service: Service<http::Request<Body>, Response = http::Response<ResBody>, Error = E>
508            + Send
509            + 'static,
510        <L::Service as Service<http::Request<Body>>>::Future: Send + 'static,
511        ResBody: http_body::Body<Data = bytes::Bytes> + Send + 'static,
512        ResBody::Error: Into<BoxError>,
513        E: Into<BoxError> + Send + 'static,
514    {
515        let layer = BoxLayer::new(
516            ServiceBuilder::new()
517                .map_response(|resp: http::Response<ResBody>| resp.map(Body::new))
518                .map_err(Into::<BoxError>::into)
519                .layer(layer),
520        );
521        Arc::make_mut(&mut self.config).request_layer = Some(layer);
522        self
523    }
524
525    pub fn with_max_decoding_message_size(mut self, limit: usize) -> Self {
526        Arc::make_mut(&mut self.config).max_decoding_message_size = Some(limit);
527        self
528    }
529
530    pub fn uri(&self) -> &http::Uri {
531        &self.config.uri
532    }
533
534    fn channel(&self) -> BoxedChannel {
535        let headers = self.config.headers.clone();
536
537        // Build the base service with headers applied at the HTTP level and the
538        // transport error mapped to BoxError for compatibility with user layers.
539        let base = BoxService::new(
540            ServiceBuilder::new()
541                .map_err(|e: tonic::transport::Error| -> BoxError { Box::new(e) })
542                .map_request(move |mut req: http::Request<Body>| {
543                    if !headers.headers().is_empty() {
544                        req.headers_mut()
545                            .extend(headers.headers().clone().into_headers());
546                    }
547                    req
548                })
549                .service(self.channel.clone()),
550        );
551
552        // Guard every response body with the idle-body watchdog, beneath any
553        // user layers so their view of the response goes through the
554        // watchdog's bridge.
555        let base = BoxService::new(WatchdogLayer::new(self.config.body_idle_timeout).layer(base));
556
557        // Apply the user's outbound request layer if present.
558        let layered = if let Some(layer) = &self.config.request_layer {
559            layer.layer(base)
560        } else {
561            base
562        };
563
564        // Map the final error to tonic::Status (a concrete type) so that
565        // downstream users of the tonic-generated clients don't run into
566        // lifetime-inference issues with async_trait and Box<dyn Error>.
567        BoxService::new(
568            ServiceBuilder::new()
569                .map_err(status_from_error)
570                .service(layered),
571        )
572    }
573
574    pub fn ledger_client(&mut self) -> LedgerServiceClient<BoxedChannel> {
575        LedgerServiceClient::new(self.channel())
576            .accept_compressed(CompressionEncoding::Zstd)
577            .pipe(|client| {
578                if let Some(limit) = self.config.max_decoding_message_size {
579                    client.max_decoding_message_size(limit)
580                } else {
581                    client
582                }
583            })
584    }
585
586    pub fn state_client(&mut self) -> StateServiceClient<BoxedChannel> {
587        StateServiceClient::new(self.channel())
588            .accept_compressed(CompressionEncoding::Zstd)
589            .pipe(|client| {
590                if let Some(limit) = self.config.max_decoding_message_size {
591                    client.max_decoding_message_size(limit)
592                } else {
593                    client
594                }
595            })
596    }
597
598    pub fn execution_client(&mut self) -> TransactionExecutionServiceClient<BoxedChannel> {
599        TransactionExecutionServiceClient::new(self.channel())
600            .accept_compressed(CompressionEncoding::Zstd)
601            .pipe(|client| {
602                if let Some(limit) = self.config.max_decoding_message_size {
603                    client.max_decoding_message_size(limit)
604                } else {
605                    client
606                }
607            })
608    }
609
610    pub fn package_client(&mut self) -> MovePackageServiceClient<BoxedChannel> {
611        MovePackageServiceClient::new(self.channel())
612            .accept_compressed(CompressionEncoding::Zstd)
613            .pipe(|client| {
614                if let Some(limit) = self.config.max_decoding_message_size {
615                    client.max_decoding_message_size(limit)
616                } else {
617                    client
618                }
619            })
620    }
621
622    pub fn signature_verification_client(
623        &mut self,
624    ) -> SignatureVerificationServiceClient<BoxedChannel> {
625        SignatureVerificationServiceClient::new(self.channel())
626            .accept_compressed(CompressionEncoding::Zstd)
627            .pipe(|client| {
628                if let Some(limit) = self.config.max_decoding_message_size {
629                    client.max_decoding_message_size(limit)
630                } else {
631                    client
632                }
633            })
634    }
635
636    pub fn subscription_client(&mut self) -> SubscriptionServiceClient<BoxedChannel> {
637        SubscriptionServiceClient::new(self.channel())
638            .accept_compressed(CompressionEncoding::Zstd)
639            .pipe(|client| {
640                if let Some(limit) = self.config.max_decoding_message_size {
641                    client.max_decoding_message_size(limit)
642                } else {
643                    client
644                }
645            })
646    }
647
648    /// Returns a client for the unstable alpha `ProofService`, which serves
649    /// Object Checkpoint State (OCS) inclusion proofs.
650    #[cfg(feature = "unstable")]
651    #[cfg_attr(doc_cfg, doc(cfg(feature = "unstable")))]
652    pub fn proof_client(&mut self) -> ProofServiceClient<BoxedChannel> {
653        ProofServiceClient::new(self.channel())
654            .accept_compressed(CompressionEncoding::Zstd)
655            .pipe(|client| {
656                if let Some(limit) = self.config.max_decoding_message_size {
657                    client.max_decoding_message_size(limit)
658                } else {
659                    client
660                }
661            })
662    }
663}
664
665/// Map a transport error to a [`tonic::Status`].
666///
667/// tonic surfaces an expired headers-phase timeout (the client's
668/// response-headers timeout, or a `grpc-timeout` deadline expiring before
669/// response headers arrive) as `Cancelled`. The gRPC code for an expired
670/// deadline is `DeadlineExceeded`, and the watchdog already uses it for
671/// body-phase expiry, so normalize before delegating to tonic's own mapping.
672fn status_from_error(error: BoxError) -> tonic::Status {
673    let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error.as_ref());
674    while let Some(err) = source {
675        // An embedded `Status` takes precedence, as in tonic's own mapping.
676        if err.is::<tonic::Status>() {
677            break;
678        }
679        if err.is::<tonic::TimeoutExpired>() {
680            return tonic::Status::deadline_exceeded(
681                "timeout expired before response headers were received",
682            );
683        }
684        source = err.source();
685    }
686    tonic::Status::from_error(error)
687}