zebra_rpc/
server.rs

1//! A JSON-RPC 1.0 & 2.0 endpoint for Zebra.
2//!
3//! This endpoint is compatible with clients that incorrectly send
4//! `"jsonrpc" = 1.0` fields in JSON-RPC 1.0 requests,
5//! such as `lightwalletd`.
6//!
7//! See the full list of
8//! [Differences between JSON-RPC 1.0 and 2.0.](https://www.simple-is-better.org/rpc/#differences-between-1-0-and-2-0)
9
10use std::{fmt, panic};
11
12use cookie::Cookie;
13use jsonrpsee::server::{middleware::rpc::RpcServiceBuilder, Server, ServerHandle};
14use tokio::task::JoinHandle;
15use tracing::*;
16
17use zebra_chain::{
18    block::MAX_BLOCK_BYTES, chain_sync_status::ChainSyncStatus, chain_tip::ChainTip,
19    parameters::Network,
20};
21use zebra_consensus::router::service_trait::BlockVerifierService;
22use zebra_network::AddressBookPeers;
23use zebra_node_services::mempool::MempoolService;
24use zebra_state::{ReadState as ReadStateService, State as StateService};
25
26use crate::{
27    config,
28    methods::{RpcImpl, RpcServer as _},
29    server::{
30        http_request_compatibility::HttpRequestMiddlewareLayer,
31        rpc_call_compatibility::FixRpcResponseMiddleware, rpc_metrics::RpcMetricsMiddleware,
32        rpc_tracing::RpcTracingMiddleware,
33    },
34};
35
36pub mod cookie;
37pub mod error;
38pub mod http_request_compatibility;
39pub mod rpc_call_compatibility;
40pub mod rpc_metrics;
41pub mod rpc_tracing;
42
43#[cfg(test)]
44mod tests;
45
46/// Zebra RPC Server
47#[derive(Clone)]
48pub struct RpcServer {
49    /// The RPC config.
50    config: config::rpc::Config,
51
52    /// The configured network.
53    network: Network,
54
55    /// Zebra's application version, with build metadata.
56    build_version: String,
57
58    /// A server handle used to shuts down the RPC server.
59    close_handle: ServerHandle,
60}
61
62impl fmt::Debug for RpcServer {
63    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
64        f.debug_struct("RpcServer")
65            .field("config", &self.config)
66            .field("network", &self.network)
67            .field("build_version", &self.build_version)
68            .field(
69                "close_handle",
70                // TODO: when it stabilises, use std::any::type_name_of_val(&self.close_handle)
71                &"ServerHandle",
72            )
73            .finish()
74    }
75}
76
77/// The message to log when logging the RPC server's listen address
78pub const OPENED_RPC_ENDPOINT_MSG: &str = "Opened RPC endpoint at ";
79
80/// The message to log when logging the lightwalletd gRPC server's listen address.
81///
82/// Deliberately distinct from [`OPENED_RPC_ENDPOINT_MSG`], and not a superstring of it:
83/// tests find listen addresses by scraping these messages out of the log, and some read
84/// several in a row and rely on their order.
85pub const OPENED_LIGHTWALLETD_ENDPOINT_MSG: &str = "Opened lightwalletd gRPC endpoint at ";
86
87type ServerTask = JoinHandle<Result<(), tower::BoxError>>;
88
89impl RpcServer {
90    /// Starts the RPC server.
91    ///
92    /// `build_version` and `user_agent` are version strings for the application,
93    /// which are used in RPC responses.
94    ///
95    /// Returns [`JoinHandle`]s for the RPC server and `sendrawtransaction` queue tasks,
96    /// and a [`RpcServer`] handle, which can be used to shut down the RPC server task.
97    ///
98    /// # Panics
99    ///
100    /// - If [`Config::listen_addr`](config::rpc::Config::listen_addr) is `None`.
101    //
102    // TODO:
103    // - replace VersionString with semver::Version, and update the tests to provide valid versions
104    #[allow(clippy::too_many_arguments)]
105    pub async fn start<
106        Mempool,
107        State,
108        ReadState,
109        Tip,
110        BlockVerifierRouter,
111        SyncStatus,
112        AddressBook,
113    >(
114        rpc: RpcImpl<Mempool, State, ReadState, Tip, AddressBook, BlockVerifierRouter, SyncStatus>,
115        conf: config::rpc::Config,
116    ) -> Result<ServerTask, tower::BoxError>
117    where
118        Mempool: MempoolService,
119        State: StateService,
120        ReadState: ReadStateService,
121        Tip: ChainTip + Clone + Send + Sync + 'static,
122        AddressBook: AddressBookPeers + Clone + Send + Sync + 'static,
123        BlockVerifierRouter: BlockVerifierService,
124        SyncStatus: ChainSyncStatus + Clone + Send + Sync + 'static,
125    {
126        let listen_addr = conf
127            .listen_addr
128            .expect("caller should make sure listen_addr is set");
129
130        // The largest RPC request is submitblock, which sends a full block
131        // as a hex string (2x MAX_BLOCK_BYTES) plus a small JSON-RPC wrapper.
132        let max_request_body_size = (MAX_BLOCK_BYTES as usize) * 2 + 1024;
133        let http_middleware_layer = if conf.enable_cookie_auth {
134            let cookie = Cookie::default();
135            cookie::write_to_disk(&cookie, &conf.cookie_dir)
136                .expect("Zebra must be able to write the auth cookie to the disk");
137            HttpRequestMiddlewareLayer::new(Some(cookie), max_request_body_size)
138        } else {
139            HttpRequestMiddlewareLayer::new(None, max_request_body_size)
140        };
141
142        let http_middleware = tower::ServiceBuilder::new().layer(http_middleware_layer);
143
144        let rpc_middleware = RpcServiceBuilder::new()
145            .rpc_logger(1024)
146            .layer_fn(FixRpcResponseMiddleware::new)
147            .layer_fn(RpcMetricsMiddleware::new)
148            .layer_fn(RpcTracingMiddleware::new);
149
150        let server = Server::builder()
151            .http_only()
152            .set_http_middleware(http_middleware)
153            .set_rpc_middleware(rpc_middleware)
154            .max_response_body_size(conf.max_response_body_size)
155            .build(listen_addr)
156            .await?;
157
158        info!("{OPENED_RPC_ENDPOINT_MSG}{}", server.local_addr()?);
159
160        Ok(tokio::spawn(async move {
161            server.start(rpc.into_rpc()).stopped().await;
162            Ok(())
163        }))
164    }
165
166    /// Shut down this RPC server, blocking the current thread.
167    ///
168    /// This method can be called from within a tokio executor without panicking.
169    /// But it is blocking, so `shutdown()` should be used instead.
170    pub fn shutdown_blocking(&self) {
171        Self::shutdown_blocking_inner(self.close_handle.clone(), self.config.clone())
172    }
173
174    /// Shut down this RPC server asynchronously.
175    /// Returns a task that completes when the server is shut down.
176    pub fn shutdown(&self) -> JoinHandle<()> {
177        let close_handle = self.close_handle.clone();
178        let config = self.config.clone();
179        let span = Span::current();
180
181        tokio::task::spawn_blocking(move || {
182            span.in_scope(|| Self::shutdown_blocking_inner(close_handle, config))
183        })
184    }
185
186    /// Shuts down this RPC server using its `close_handle`.
187    ///
188    /// See `shutdown_blocking()` for details.
189    fn shutdown_blocking_inner(close_handle: ServerHandle, config: config::rpc::Config) {
190        // The server is a blocking task, so it can't run inside a tokio thread.
191        // See the note at wait_on_server.
192        let span = Span::current();
193        let wait_on_shutdown = move || {
194            span.in_scope(|| {
195                if config.enable_cookie_auth {
196                    if let Err(err) = cookie::remove_from_disk(&config.cookie_dir) {
197                        warn!(
198                            ?err,
199                            "unexpectedly could not remove the rpc auth cookie from the disk"
200                        )
201                    }
202                }
203
204                info!("Stopping RPC server");
205                let _ = close_handle.stop();
206                debug!("Stopped RPC server");
207            })
208        };
209
210        let span = Span::current();
211        let thread_handle = std::thread::spawn(wait_on_shutdown);
212
213        // Propagate panics from the inner std::thread to the outer tokio blocking task
214        span.in_scope(|| match thread_handle.join() {
215            Ok(()) => (),
216            Err(panic_object) => panic::resume_unwind(panic_object),
217        })
218    }
219}
220
221impl Drop for RpcServer {
222    fn drop(&mut self) {
223        // Block on shutting down, propagating panics.
224        // This can take around 150 seconds.
225        //
226        // Without this shutdown, Zebra's RPC unit tests sometimes crashed with memory errors.
227        self.shutdown_blocking();
228    }
229}
zebra_rpc/server.rs

zebra_rpc/
server.rs
