zebra_node_services/
rpc_client.rs

1//! A client for calling Zebra's JSON-RPC methods.
2//!
3//! Used in the rpc sync scanning functionality and in various tests and tools.
4
5use std::{net::SocketAddr, time::Duration};
6
7use reqwest::Client;
8
9use crate::BoxError;
10
11/// The default timeout for RPC requests.
12///
13/// This is a safety net to prevent RPC calls from hanging indefinitely
14/// when a server is alive but unresponsive.
15const RPC_REQUEST_TIMEOUT: Duration = Duration::from_secs(180);
16
17/// An HTTP client for making JSON-RPC requests.
18#[derive(Clone, Debug)]
19pub struct RpcRequestClient {
20    client: Client,
21    rpc_address: SocketAddr,
22}
23
24impl RpcRequestClient {
25    /// Creates a new RPC request client with the default timeout.
26    pub fn new(rpc_address: SocketAddr) -> Self {
27        Self::new_with_timeout(rpc_address, RPC_REQUEST_TIMEOUT)
28    }
29
30    /// Creates a new RPC request client with a custom request timeout.
31    ///
32    /// Use [`RpcRequestClient::new()`] for the default timeout.
33    pub fn new_with_timeout(rpc_address: SocketAddr, timeout: Duration) -> Self {
34        Self {
35            client: Client::builder()
36                .timeout(timeout)
37                .build()
38                .expect("reqwest::Client build should not fail when only setting timeout"),
39            rpc_address,
40        }
41    }
42
43    /// Builds rpc request
44    pub async fn call(
45        &self,
46        method: impl AsRef<str>,
47        params: impl AsRef<str>,
48    ) -> reqwest::Result<reqwest::Response> {
49        let method = method.as_ref();
50        let params = params.as_ref();
51
52        self.client
53            .post(format!("http://{}", self.rpc_address))
54            .body(format!(
55                r#"{{"jsonrpc": "2.0", "method": "{method}", "params": {params}, "id":123 }}"#
56            ))
57            .header("Content-Type", "application/json")
58            .send()
59            .await
60    }
61
62    /// Builds rpc request with a variable `content-type`.
63    pub async fn call_with_content_type(
64        &self,
65        method: impl AsRef<str>,
66        params: impl AsRef<str>,
67        content_type: String,
68    ) -> reqwest::Result<reqwest::Response> {
69        let method = method.as_ref();
70        let params = params.as_ref();
71
72        self.client
73            .post(format!("http://{}", self.rpc_address))
74            .body(format!(
75                r#"{{"jsonrpc": "2.0", "method": "{method}", "params": {params}, "id":123 }}"#
76            ))
77            .header("Content-Type", content_type)
78            .send()
79            .await
80    }
81
82    /// Builds rpc request with no content type.
83    pub async fn call_with_no_content_type(
84        &self,
85        method: impl AsRef<str>,
86        params: impl AsRef<str>,
87    ) -> reqwest::Result<reqwest::Response> {
88        let method = method.as_ref();
89        let params = params.as_ref();
90
91        self.client
92            .post(format!("http://{}", self.rpc_address))
93            .body(format!(
94                r#"{{"jsonrpc": "2.0", "method": "{method}", "params": {params}, "id":123 }}"#
95            ))
96            .send()
97            .await
98    }
99
100    /// Builds rpc request and gets text from response
101    pub async fn text_from_call(
102        &self,
103        method: impl AsRef<str>,
104        params: impl AsRef<str>,
105    ) -> reqwest::Result<String> {
106        self.call(method, params).await?.text().await
107    }
108
109    /// Builds an RPC request, awaits its response, and attempts to deserialize
110    /// it to the expected result type.
111    ///
112    /// Returns Ok with json result from response if successful.
113    /// Returns an error if the call or result deserialization fail.
114    pub async fn json_result_from_call<T: serde::de::DeserializeOwned>(
115        &self,
116        method: impl AsRef<str>,
117        params: impl AsRef<str>,
118    ) -> std::result::Result<T, BoxError> {
119        Self::json_result_from_response_text(&self.text_from_call(method, params).await?)
120    }
121
122    /// Accepts response text from an RPC call
123    /// Returns `Ok` with a deserialized `result` value in the expected type, or an error report.
124    fn json_result_from_response_text<T: serde::de::DeserializeOwned>(
125        response_text: &str,
126    ) -> std::result::Result<T, BoxError> {
127        let output: jsonrpsee_types::Response<serde_json::Value> =
128            serde_json::from_str(response_text)?;
129        match output.payload {
130            jsonrpsee_types::ResponsePayload::Success(success) => {
131                Ok(serde_json::from_value(success.into_owned())?)
132            }
133            jsonrpsee_types::ResponsePayload::Error(failure) => Err(failure.to_string().into()),
134        }
135    }
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141    use std::sync::{Arc, Mutex};
142
143    /// Proves that `RpcRequestClient` times out instead of hanging indefinitely
144    /// when a server accepts a TCP connection but never sends a response.
145    #[tokio::test]
146    async fn rpc_client_timeout_on_unresponsive_server() {
147        let listener =
148            std::net::TcpListener::bind("127.0.0.1:0").expect("should bind to localhost");
149        let addr = listener.local_addr().expect("should have a local address");
150
151        // Hold the accepted stream so the connection stays open until the test ends.
152        let held_stream = Arc::new(Mutex::new(None));
153        let held_stream_clone = held_stream.clone();
154
155        // Accept the connection but never respond.
156        let accept_thread = std::thread::spawn(move || {
157            let (stream, _peer_addr) = listener.accept().expect("should accept a connection");
158            *held_stream_clone.lock().expect("lock poisoned") = Some(stream);
159        });
160
161        let short_timeout = Duration::from_secs(2);
162        let client = RpcRequestClient::new_with_timeout(addr, short_timeout);
163
164        // Outer timeout is a safety net — should never fire.
165        let result = tokio::time::timeout(
166            Duration::from_secs(30),
167            client.text_from_call("getinfo", "[]"),
168        )
169        .await;
170
171        let inner_result =
172            result.expect("outer safety timeout should not fire; client timeout should fire first");
173
174        let err =
175            inner_result.expect_err("request to unresponsive server should fail with timeout");
176        assert!(err.is_timeout(), "error should be a timeout, got: {err}");
177
178        drop(held_stream);
179        accept_thread
180            .join()
181            .expect("accept thread should exit cleanly");
182    }
183}
zebra_node_services/rpc_client.rs

zebra_node_services/
rpc_client.rs
