sui_rpc_api/
lib.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::convert::Infallible;
5use std::sync::Arc;
6
7use reader::StateReader;
8use subscription::SubscriptionServiceHandle;
9use sui_http::middleware::callback::CallbackLayer;
10use sui_types::storage::RpcStateReader;
11use sui_types::transaction_executor::ProposerSelector;
12use sui_types::transaction_executor::TransactionExecutor;
13use tap::Pipe;
14use tonic::server::NamedService;
15use tower::Service;
16
17pub mod client;
18mod config;
19mod error;
20pub mod grpc;
21pub mod ledger_history;
22mod metrics;
23pub mod read_mask_defaults;
24mod reader;
25mod response;
26mod service;
27pub mod subscription;
28
29pub use client::Client;
30pub use config::Config;
31pub use error::{
32    CheckpointNotFoundError, ErrorDetails, ErrorReason, ObjectNotFoundError, Result, RpcError,
33};
34pub use metrics::{
35    GrpcMethodAllowlist, RpcMetrics, RpcMetricsMakeCallbackHandler,
36    grpc_method_paths_from_file_descriptor_sets,
37};
38pub use reader::TransactionNotFoundError;
39pub use sui_rpc::proto;
40
41#[derive(Clone)]
42pub struct ServerVersion {
43    pub bin: &'static str,
44    pub version: &'static str,
45}
46
47impl ServerVersion {
48    pub fn new(bin: &'static str, version: &'static str) -> Self {
49        Self { bin, version }
50    }
51}
52
53impl std::fmt::Display for ServerVersion {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        f.write_str(self.bin)?;
56        f.write_str("/")?;
57        f.write_str(self.version)
58    }
59}
60
61#[derive(Clone)]
62pub struct RpcService {
63    reader: StateReader,
64    executor: Option<Arc<dyn TransactionExecutor>>,
65    proposer_selector: Option<Arc<dyn ProposerSelector>>,
66    subscription_service_handle: Option<SubscriptionServiceHandle>,
67    chain_id: sui_types::digests::ChainIdentifier,
68    server_version: Option<ServerVersion>,
69    metrics: Option<Arc<RpcMetrics>>,
70    pub(crate) list_metrics: Option<Arc<metrics::ListApiMetrics>>,
71    config: Config,
72    extra_routes: axum::Router,
73    extra_service_names: Vec<&'static str>,
74    extra_file_descriptor_sets: Vec<&'static [u8]>,
75}
76
77impl RpcService {
78    pub fn new(reader: Arc<dyn RpcStateReader>) -> Self {
79        let chain_id = reader.get_chain_identifier().unwrap();
80        Self {
81            reader: StateReader::new(reader),
82            executor: None,
83            proposer_selector: None,
84            subscription_service_handle: None,
85            chain_id,
86            server_version: None,
87            metrics: None,
88            list_metrics: None,
89            config: Config::default(),
90            extra_routes: axum::Router::new(),
91            extra_service_names: Vec::new(),
92            extra_file_descriptor_sets: Vec::new(),
93        }
94    }
95
96    pub fn with_server_version(&mut self, server_version: ServerVersion) -> &mut Self {
97        self.server_version = Some(server_version);
98        self
99    }
100
101    pub fn with_config(&mut self, config: Config) {
102        self.config = config;
103    }
104
105    pub fn with_executor(&mut self, executor: Arc<dyn TransactionExecutor + Send + Sync>) {
106        self.executor = Some(executor);
107    }
108
109    pub fn with_proposer_selector(&mut self, proposer_selector: Arc<dyn ProposerSelector>) {
110        self.proposer_selector = Some(proposer_selector);
111    }
112
113    pub fn with_subscription_service(
114        &mut self,
115        subscription_service_handle: SubscriptionServiceHandle,
116    ) {
117        self.subscription_service_handle = Some(subscription_service_handle);
118    }
119
120    pub fn with_metrics(&mut self, registry: &prometheus::Registry) {
121        self.metrics = Some(Arc::new(RpcMetrics::new(registry)));
122        self.list_metrics = Some(Arc::new(metrics::ListApiMetrics::new(registry)));
123    }
124
125    pub fn with_custom_service<S>(&mut self, svc: S)
126    where
127        S: Service<
128                axum::extract::Request,
129                Response: axum::response::IntoResponse,
130                Error = Infallible,
131            > + NamedService
132            + Clone
133            + Send
134            + Sync
135            + 'static,
136        S::Future: Send + 'static,
137        S::Error: Into<grpc::BoxError> + Send,
138    {
139        self.extra_service_names.push(S::NAME);
140        self.extra_routes = std::mem::take(&mut self.extra_routes)
141            .route_service(&format!("/{}/{{*rest}}", S::NAME), svc);
142    }
143
144    pub fn with_file_descriptor_set(&mut self, encoded_fds: &'static [u8]) {
145        self.extra_file_descriptor_sets.push(encoded_fds);
146    }
147
148    pub fn chain_id(&self) -> sui_types::digests::ChainIdentifier {
149        self.chain_id
150    }
151
152    pub fn server_version(&self) -> Option<&ServerVersion> {
153        self.server_version.as_ref()
154    }
155
156    pub async fn into_router(mut self) -> axum::Router {
157        let metrics = self.metrics.clone();
158        let extra_routes = std::mem::take(&mut self.extra_routes);
159        let extra_service_names = std::mem::take(&mut self.extra_service_names);
160
161        // Single source of truth for every encoded FileDescriptorSet that
162        // backs a gRPC service mounted below. Consumed by the reflection
163        // services, the metrics allowlist, and the request-log layer so they
164        // cannot drift out of sync.
165        let built_in_file_descriptor_sets: [&[u8]; 5] = [
166            sui_rpc::proto::google::protobuf::FILE_DESCRIPTOR_SET,
167            sui_rpc::proto::google::rpc::FILE_DESCRIPTOR_SET,
168            sui_rpc::proto::sui::rpc::v2::FILE_DESCRIPTOR_SET,
169            sui_rpc::proto::sui::rpc::v2alpha::FILE_DESCRIPTOR_SET,
170            tonic_health::pb::FILE_DESCRIPTOR_SET,
171        ];
172        let file_descriptor_sets: Vec<&[u8]> = built_in_file_descriptor_sets
173            .into_iter()
174            .chain(std::mem::take(&mut self.extra_file_descriptor_sets))
175            .collect();
176
177        // Allowlist of `/Service/Method` paths used by the metrics middleware
178        // to bound prometheus label cardinality.
179        let grpc_method_allowlist = Arc::new(
180            metrics::grpc_method_paths_from_file_descriptor_sets(&file_descriptor_sets)
181                .expect("registered FileDescriptorSet bytes must be valid protobuf"),
182        );
183
184        let request_log =
185            mysten_network::request_log::GrpcRequestLogLayer::from_encoded_file_descriptor_sets(
186                file_descriptor_sets.iter().copied(),
187            )
188            .unwrap_or_else(|e| {
189                // Extra sets registered by embedders may not merge cleanly (e.g. missing
190                // imports). Reflection and metrics tolerate that, so don't fail startup —
191                // capture just won't decode those extra services.
192                tracing::warn!(
193                    "request-log descriptor pool falling back to built-in file descriptor sets: {e}"
194                );
195                mysten_network::request_log::GrpcRequestLogLayer::from_encoded_file_descriptor_sets(
196                    built_in_file_descriptor_sets,
197                )
198                .expect("built-in FileDescriptorSet bytes must be valid protobuf")
199            });
200
201        let router = {
202            let ledger_service =
203                sui_rpc::proto::sui::rpc::v2::ledger_service_server::LedgerServiceServer::new(
204                    self.clone(),
205                )
206                .send_compressed(tonic::codec::CompressionEncoding::Zstd);
207            let proof_service_v2alpha =
208                sui_rpc::proto::sui::rpc::v2alpha::proof_service_server::ProofServiceServer::new(
209                    self.clone(),
210                )
211                .send_compressed(tonic::codec::CompressionEncoding::Zstd);
212            let transaction_execution_service = sui_rpc::proto::sui::rpc::v2::transaction_execution_service_server::TransactionExecutionServiceServer::new(self.clone())
213                .send_compressed(tonic::codec::CompressionEncoding::Zstd);
214            let state_service =
215                sui_rpc::proto::sui::rpc::v2::state_service_server::StateServiceServer::new(
216                    self.clone(),
217                )
218                .send_compressed(tonic::codec::CompressionEncoding::Zstd);
219            let signature_verification_service = sui_rpc::proto::sui::rpc::v2::signature_verification_service_server::SignatureVerificationServiceServer::new(self.clone())
220                .send_compressed(tonic::codec::CompressionEncoding::Zstd);
221            let move_package_service = sui_rpc::proto::sui::rpc::v2::move_package_service_server::MovePackageServiceServer::new(self.clone())
222                .send_compressed(tonic::codec::CompressionEncoding::Zstd);
223            let name_service =
224                sui_rpc::proto::sui::rpc::v2::name_service_server::NameServiceServer::new(
225                    self.clone(),
226                )
227                .send_compressed(tonic::codec::CompressionEncoding::Zstd);
228
229            let (health_reporter, health_service) = tonic_health::server::health_reporter();
230
231            let mut reflection_v1_builder = tonic_reflection::server::Builder::configure();
232            let mut reflection_v1alpha_builder = tonic_reflection::server::Builder::configure();
233            for fds in &file_descriptor_sets {
234                reflection_v1_builder =
235                    reflection_v1_builder.register_encoded_file_descriptor_set(fds);
236                reflection_v1alpha_builder =
237                    reflection_v1alpha_builder.register_encoded_file_descriptor_set(fds);
238            }
239
240            let reflection_v1 = reflection_v1_builder.build_v1().unwrap();
241            let reflection_v1alpha = reflection_v1alpha_builder.build_v1alpha().unwrap();
242
243            fn service_name<S: tonic::server::NamedService>(_service: &S) -> &'static str {
244                S::NAME
245            }
246
247            for service_name in [
248                service_name(&ledger_service),
249                service_name(&transaction_execution_service),
250                service_name(&state_service),
251                service_name(&signature_verification_service),
252                service_name(&move_package_service),
253                service_name(&name_service),
254                service_name(&proof_service_v2alpha),
255                service_name(&reflection_v1),
256                service_name(&reflection_v1alpha),
257            ] {
258                health_reporter
259                    .set_service_status(service_name, tonic_health::ServingStatus::Serving)
260                    .await;
261            }
262
263            let mut services = grpc::Services::new()
264                .timeout(self.config.grpc_timeout())
265                // V2
266                .add_service(ledger_service)
267                .add_service(transaction_execution_service)
268                .add_service(state_service)
269                .add_service(signature_verification_service)
270                .add_service(move_package_service)
271                .add_service(name_service)
272                // V2alpha
273                .add_service(proof_service_v2alpha)
274                // Reflection
275                .add_service(reflection_v1)
276                .add_service(reflection_v1alpha);
277
278            if self.subscription_service_handle.is_some() {
279                let subscription_service =
280sui_rpc::proto::sui::rpc::v2::subscription_service_server::SubscriptionServiceServer::new(self.clone());
281                health_reporter
282                    .set_service_status(
283                        service_name(&subscription_service),
284                        tonic_health::ServingStatus::Serving,
285                    )
286                    .await;
287
288                services = services.add_service(subscription_service);
289            }
290
291            for name in &extra_service_names {
292                health_reporter
293                    .set_service_status(*name, tonic_health::ServingStatus::Serving)
294                    .await;
295            }
296
297            services
298                .merge_router(extra_routes)
299                .add_service(health_service)
300                .into_router(request_log)
301        };
302
303        let health_endpoint = axum::Router::new()
304            .route("/health", axum::routing::get(service::health::health))
305            .with_state(self.clone());
306
307        router
308            .merge(health_endpoint)
309            .layer(axum::middleware::map_response_with_state(
310                self,
311                response::append_info_headers,
312            ))
313            .pipe(|router| {
314                if let Some(metrics) = metrics {
315                    router.layer(CallbackLayer::new(
316                        metrics::RpcMetricsMakeCallbackHandler::with_grpc_method_allowlist(
317                            metrics,
318                            grpc_method_allowlist,
319                        ),
320                    ))
321                } else {
322                    router
323                }
324            })
325    }
326
327    pub async fn start_service(self, socket_address: std::net::SocketAddr) {
328        let listener = tokio::net::TcpListener::bind(socket_address).await.unwrap();
329        axum::serve(listener, self.into_router().await)
330            .await
331            .unwrap();
332    }
333}
334
335#[derive(Debug, Copy, Clone, serde::Serialize, serde::Deserialize)]
336#[serde(rename_all = "lowercase")]
337pub enum Direction {
338    Ascending,
339    Descending,
340}
341
342impl Direction {
343    pub fn is_descending(self) -> bool {
344        matches!(self, Self::Descending)
345    }
346}
347
348#[cfg(test)]
349mod tests {
350    /// The request-log layer's descriptor pool is built from these sets at server startup with an
351    /// `expect`, so they must always merge into one valid pool.
352    #[test]
353    fn request_log_pool_builds_from_registered_file_descriptor_sets() {
354        mysten_network::request_log::GrpcRequestLogLayer::from_encoded_file_descriptor_sets([
355            sui_rpc::proto::google::protobuf::FILE_DESCRIPTOR_SET,
356            sui_rpc::proto::google::rpc::FILE_DESCRIPTOR_SET,
357            sui_rpc::proto::sui::rpc::v2::FILE_DESCRIPTOR_SET,
358            sui_rpc::proto::sui::rpc::v2alpha::FILE_DESCRIPTOR_SET,
359            tonic_health::pb::FILE_DESCRIPTOR_SET,
360        ])
361        .unwrap();
362    }
363}
sui_rpc_api/lib.rs

sui_rpc_api/
lib.rs
