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

sui_rpc_api/
lib.rs
