iota_json_rpc/
lib.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5// Raise rustc's query-depth limit for monomorphizing deeply-nested generic
6// futures. The JSON-RPC handler routes requests through axum → orchestrator
7// → `submit_with_checkpoint_race` (a `tokio::select!` over the driver and
8// checkpoint-inclusion arms) → transaction_driver → effects_certifier →
9// `FuturesUnordered` of per-validator queries → safe_client RPC. Each
10// `.await` and combinator adds one nested anonymous Future type, and when
11// `iota_metrics::spawn_monitored_task!` wraps the resulting future, computing
12// its layout walks the entire chain — overshooting the default limit of 128
13// by ~2 today. 256 leaves headroom; if a future change pushes us past it,
14// that is the signal to box-pin a major arm rather than bump again. See
15// `iota-indexer/src/lib.rs` for the same pattern, same reason.
16#![recursion_limit = "256"]
17
18use std::{env, net::SocketAddr, str::FromStr, sync::Arc};
19
20use axum::{
21    body::Body,
22    routing::{get, post},
23};
24pub use balance_changes::*;
25use hyper::{
26    Method, Request,
27    header::{HeaderName, HeaderValue},
28};
29pub use iota_config::node::ServerType;
30use iota_json_rpc_api::{
31    CLIENT_SDK_TYPE_HEADER, CLIENT_SDK_VERSION_HEADER, CLIENT_TARGET_API_VERSION_HEADER,
32};
33use iota_open_rpc::{Module, Project};
34use iota_traffic_controller::TrafficController;
35use iota_types::traffic_control::PolicyConfig;
36use jsonrpsee::{Extensions, RpcModule, types::ErrorObjectOwned};
37pub use object_changes::*;
38use prometheus_filtered::Registry;
39use tokio::runtime::Handle;
40use tokio_util::sync::CancellationToken;
41use tower_http::{
42    cors::{AllowOrigin, CorsLayer},
43    trace::TraceLayer,
44};
45use tracing::{debug, info};
46
47use crate::{
48    axum_router::{json_rpc_handler, ws::ws_json_rpc_upgrade},
49    error::Error,
50    metrics::MetricsLogger,
51    routing_layer::RpcRouter,
52};
53
54pub mod authority_state;
55pub mod axum_router;
56mod balance_changes;
57pub mod coin_api;
58pub mod error;
59pub mod governance_api;
60pub mod indexer_api;
61pub mod logger;
62mod metrics;
63pub mod move_utils;
64mod object_changes;
65pub mod read_api;
66mod routing_layer;
67pub mod transaction_builder_api;
68pub mod transaction_execution_api;
69
70pub const APP_NAME_HEADER: &str = "app-name";
71
72pub const MAX_REQUEST_SIZE: u32 = 2 << 30;
73
74pub struct JsonRpcServerBuilder {
75    module: RpcModule<()>,
76    rpc_doc: Project,
77    registry: Registry,
78    traffic_controller: Option<Arc<TrafficController>>,
79    policy_config: Option<PolicyConfig>,
80}
81
82pub fn iota_rpc_doc(version: &str) -> Project {
83    Project::new(
84        version,
85        "IOTA JSON-RPC",
86        "IOTA JSON-RPC API for interaction with IOTA full node or indexer. Make RPC calls using https://api.NETWORK.iota.cafe:443 (or https://indexer.NETWORK.iota.cafe:443 for the indexer), where NETWORK is the network you want to use (testnet, devnet, mainnet). By default, local networks use port 9000 (or 9124 for the indexer).",
87        "IOTA Foundation",
88        "https://iota.org",
89        "info@iota.org",
90        "Apache-2.0",
91        "https://raw.githubusercontent.com/iotaledger/iota/main/LICENSE",
92    )
93}
94
95impl JsonRpcServerBuilder {
96    pub fn new(
97        version: &str,
98        prometheus_registry: &Registry,
99        traffic_controller: Option<Arc<TrafficController>>,
100        policy_config: Option<PolicyConfig>,
101    ) -> Self {
102        Self {
103            module: RpcModule::new(()),
104            rpc_doc: iota_rpc_doc(version),
105            registry: prometheus_registry.clone(),
106            traffic_controller,
107            policy_config,
108        }
109    }
110
111    pub fn register_module<T: IotaRpcModule>(&mut self, module: T) -> Result<(), Error> {
112        self.rpc_doc.add_module(T::rpc_doc_module());
113        Ok(self.module.merge(module.rpc())?)
114    }
115
116    fn cors() -> Result<CorsLayer, Error> {
117        let acl = match env::var("ACCESS_CONTROL_ALLOW_ORIGIN") {
118            Ok(value) => {
119                let allow_hosts = value
120                    .split(',')
121                    .map(HeaderValue::from_str)
122                    .collect::<Result<Vec<_>, _>>()?;
123                AllowOrigin::list(allow_hosts)
124            }
125            _ => AllowOrigin::any(),
126        };
127        info!(?acl);
128
129        let cors = CorsLayer::new()
130            // Allow `POST` when accessing the resource
131            .allow_methods([Method::POST])
132            // Allow requests from any origin
133            .allow_origin(acl)
134            .allow_headers([
135                hyper::header::CONTENT_TYPE,
136                HeaderName::from_static(CLIENT_SDK_TYPE_HEADER),
137                HeaderName::from_static(CLIENT_SDK_VERSION_HEADER),
138                HeaderName::from_static(CLIENT_TARGET_API_VERSION_HEADER),
139                HeaderName::from_static(APP_NAME_HEADER),
140            ]);
141        Ok(cors)
142    }
143
144    fn trace_layer() -> TraceLayer<
145        tower_http::classify::SharedClassifier<tower_http::classify::ServerErrorsAsFailures>,
146        impl tower_http::trace::MakeSpan<Body> + Clone,
147    > {
148        TraceLayer::new_for_http().make_span_with(|request: &Request<Body>| {
149            let request_id = request
150                .headers()
151                .get("x-req-id")
152                .and_then(|v| v.to_str().ok())
153                .map(tracing::field::display);
154
155            tracing::info_span!("json-rpc-request", "x-req-id" = request_id)
156        })
157    }
158
159    pub async fn to_router(&self, server_type: ServerType) -> Result<axum::Router, Error> {
160        let routing = self.rpc_doc.method_routing.clone();
161
162        let disable_routing = env::var("DISABLE_BACKWARD_COMPATIBILITY")
163            .ok()
164            .and_then(|v| bool::from_str(&v).ok())
165            .unwrap_or_default();
166        info!(
167            "Compatibility method routing {}.",
168            if disable_routing {
169                "disabled"
170            } else {
171                "enabled"
172            }
173        );
174        let rpc_router = RpcRouter::new(routing, disable_routing);
175
176        let rpc_docs = self.rpc_doc.clone();
177        let mut module = self.module.clone();
178        module.register_method("rpc.discover", move |_, _, _| {
179            Result::<_, ErrorObjectOwned>::Ok(rpc_docs.clone())
180        })?;
181        let methods_names = module.method_names().collect::<Vec<_>>();
182
183        let metrics_logger = MetricsLogger::new(&self.registry, &methods_names);
184
185        let middleware = tower::ServiceBuilder::new()
186            .layer(Self::trace_layer())
187            .layer(Self::cors()?);
188
189        let service = crate::axum_router::JsonRpcService::new(
190            module.into(),
191            rpc_router,
192            metrics_logger,
193            self.traffic_controller.clone(),
194            self.policy_config.clone(),
195            Extensions::new(),
196        );
197
198        let mut router = axum::Router::new();
199
200        match server_type {
201            ServerType::WebSocket => {
202                router = router
203                    .route("/", get(ws_json_rpc_upgrade))
204                    .route("/subscribe", get(ws_json_rpc_upgrade));
205            }
206            ServerType::Http => {
207                router = router
208                    .route("/", post(json_rpc_handler))
209                    .route("/json-rpc", post(json_rpc_handler))
210                    .route("/public", post(json_rpc_handler));
211            }
212            ServerType::Both => {
213                router = router
214                    .route("/", post(json_rpc_handler))
215                    .route("/", get(ws_json_rpc_upgrade))
216                    .route("/subscribe", get(ws_json_rpc_upgrade))
217                    .route("/json-rpc", post(json_rpc_handler))
218                    .route("/public", post(json_rpc_handler));
219            }
220        }
221
222        let app = router.with_state(service).layer(middleware);
223
224        info!("Available JSON-RPC methods : {methods_names:?}");
225
226        Ok(app)
227    }
228
229    pub async fn start(
230        self,
231        listen_address: SocketAddr,
232        custom_runtime: Option<Handle>,
233        server_type: ServerType,
234        cancel: Option<CancellationToken>,
235    ) -> Result<ServerHandle, Error> {
236        let app = self.to_router(server_type).await?;
237
238        let listener = tokio::net::TcpListener::bind(listen_address)
239            .await
240            .map_err(|e| {
241                Error::Unexpected(format!("invalid listen address {listen_address}: {e}"))
242            })?;
243
244        let addr = listener.local_addr().map_err(|e| {
245            Error::Unexpected(format!("invalid listen address {listen_address}: {e}"))
246        })?;
247
248        let serve = axum::serve(
249            listener,
250            app.into_make_service_with_connect_info::<SocketAddr>(),
251        );
252        let shutdown = cancel.clone().map(|token| token.cancelled_owned());
253        let run_server = async move {
254            if let Some(shutdown) = shutdown {
255                serve
256                    .with_graceful_shutdown(shutdown)
257                    .await
258                    .inspect(|_| info!("Shutting down IOTA JSON-RPC server"))
259                    .unwrap()
260            } else {
261                serve.await.unwrap()
262            };
263            if let Some(token) = cancel {
264                token.cancel();
265            }
266        };
267
268        let handle = if let Some(custom_runtime) = custom_runtime {
269            debug!("Spawning server with custom runtime");
270            custom_runtime.spawn(run_server)
271        } else {
272            tokio::spawn(run_server)
273        };
274
275        let handle = ServerHandle {
276            handle: ServerHandleInner::Axum(handle),
277        };
278        info!(local_addr =? addr, "IOTA JSON-RPC server listening on {addr}");
279        Ok(handle)
280    }
281}
282
283pub struct ServerHandle {
284    handle: ServerHandleInner,
285}
286
287impl ServerHandle {
288    pub async fn stopped(self) {
289        match self.handle {
290            ServerHandleInner::Axum(handle) => handle.await.unwrap(),
291        }
292    }
293}
294
295enum ServerHandleInner {
296    Axum(tokio::task::JoinHandle<()>),
297}
298
299pub trait IotaRpcModule
300where
301    Self: Sized,
302{
303    fn rpc(self) -> RpcModule<Self>;
304    fn rpc_doc_module() -> Module;
305}
iota_json_rpc/lib.rs

iota_json_rpc/
lib.rs
