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::{
6    net::SocketAddr,
7    sync::atomic::{AtomicU64, Ordering},
8    time::Duration,
9};
10
11use jsonrpsee_types::Id;
12use reqwest::Client;
13
14use crate::BoxError;
15
16/// The default timeout for RPC requests.
17///
18/// This is a safety net to prevent RPC calls from hanging indefinitely
19/// when a server is alive but unresponsive.
20const RPC_REQUEST_TIMEOUT: Duration = Duration::from_secs(180);
21
22/// A process-wide, monotonically increasing counter for JSON-RPC request ids, so
23/// each request carries a distinct id that its response can be correlated against.
24static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1);
25
26/// Returns the next distinct JSON-RPC request id.
27fn next_request_id() -> u64 {
28    NEXT_REQUEST_ID.fetch_add(1, Ordering::Relaxed)
29}
30
31/// An HTTP client for making JSON-RPC requests.
32#[derive(Clone, Debug)]
33pub struct RpcRequestClient {
34    client: Client,
35    rpc_address: SocketAddr,
36}
37
38impl RpcRequestClient {
39    /// Creates a new RPC request client with the default timeout.
40    pub fn new(rpc_address: SocketAddr) -> Self {
41        Self::new_with_timeout(rpc_address, RPC_REQUEST_TIMEOUT)
42    }
43
44    /// Creates a new RPC request client with a custom request timeout.
45    ///
46    /// Use [`RpcRequestClient::new()`] for the default timeout.
47    pub fn new_with_timeout(rpc_address: SocketAddr, timeout: Duration) -> Self {
48        Self {
49            client: Client::builder()
50                .timeout(timeout)
51                .build()
52                .expect("reqwest::Client build should not fail when only setting timeout"),
53            rpc_address,
54        }
55    }
56
57    /// Builds rpc request
58    pub async fn call(
59        &self,
60        method: impl AsRef<str>,
61        params: impl AsRef<str>,
62    ) -> reqwest::Result<reqwest::Response> {
63        self.call_with_id(method.as_ref(), params.as_ref(), next_request_id())
64            .await
65    }
66
67    /// Builds an rpc request with the given request `id` and an `application/json`
68    /// content type.
69    async fn call_with_id(
70        &self,
71        method: &str,
72        params: &str,
73        id: u64,
74    ) -> reqwest::Result<reqwest::Response> {
75        self.client
76            .post(format!("http://{}", self.rpc_address))
77            .body(format!(
78                r#"{{"jsonrpc": "2.0", "method": "{method}", "params": {params}, "id":{id} }}"#
79            ))
80            .header("Content-Type", "application/json")
81            .send()
82            .await
83    }
84
85    /// Builds rpc request with a variable `content-type`.
86    pub async fn call_with_content_type(
87        &self,
88        method: impl AsRef<str>,
89        params: impl AsRef<str>,
90        content_type: String,
91    ) -> reqwest::Result<reqwest::Response> {
92        let method = method.as_ref();
93        let params = params.as_ref();
94        let id = next_request_id();
95
96        self.client
97            .post(format!("http://{}", self.rpc_address))
98            .body(format!(
99                r#"{{"jsonrpc": "2.0", "method": "{method}", "params": {params}, "id":{id} }}"#
100            ))
101            .header("Content-Type", content_type)
102            .send()
103            .await
104    }
105
106    /// Builds rpc request with no content type.
107    pub async fn call_with_no_content_type(
108        &self,
109        method: impl AsRef<str>,
110        params: impl AsRef<str>,
111    ) -> reqwest::Result<reqwest::Response> {
112        let method = method.as_ref();
113        let params = params.as_ref();
114        let id = next_request_id();
115
116        self.client
117            .post(format!("http://{}", self.rpc_address))
118            .body(format!(
119                r#"{{"jsonrpc": "2.0", "method": "{method}", "params": {params}, "id":{id} }}"#
120            ))
121            .send()
122            .await
123    }
124
125    /// Builds rpc request and gets text from response
126    pub async fn text_from_call(
127        &self,
128        method: impl AsRef<str>,
129        params: impl AsRef<str>,
130    ) -> reqwest::Result<String> {
131        self.call(method, params).await?.text().await
132    }
133
134    /// Builds an RPC request, awaits its response, and attempts to deserialize
135    /// it to the expected result type.
136    ///
137    /// Returns Ok with json result from response if successful.
138    /// Returns an error if the call fails, the response is missing the `jsonrpc`
139    /// version, a successful response's `id` doesn't match the request, or result
140    /// deserialization fails.
141    pub async fn json_result_from_call<T: serde::de::DeserializeOwned>(
142        &self,
143        method: impl AsRef<str>,
144        params: impl AsRef<str>,
145    ) -> std::result::Result<T, BoxError> {
146        let id = next_request_id();
147        let response_text = self
148            .call_with_id(method.as_ref(), params.as_ref(), id)
149            .await?
150            .text()
151            .await?;
152
153        Self::json_result_from_response_text(&response_text, Id::Number(id))
154    }
155
156    /// Accepts response text from an RPC call and the id of the request it answers.
157    ///
158    /// Returns `Ok` with a deserialized `result` value in the expected type, or an
159    /// error report. Rejects a response whose `jsonrpc` version is absent (a
160    /// non-`"2.0"` version already fails to deserialize), and a *successful*
161    /// response whose `id` does not match `expected_id`. Error payloads are
162    /// returned as-is (a null `id` is valid on some JSON-RPC errors).
163    fn json_result_from_response_text<T: serde::de::DeserializeOwned>(
164        response_text: &str,
165        expected_id: Id<'_>,
166    ) -> std::result::Result<T, BoxError> {
167        let output: jsonrpsee_types::Response<serde_json::Value> =
168            serde_json::from_str(response_text)?;
169
170        // A wrong `jsonrpc` version already fails to deserialize above (the field
171        // only accepts "2.0"), so here we only need to reject a missing version.
172        if output.jsonrpc.is_none() {
173            return Err("JSON-RPC response is missing the \"jsonrpc\" version field".into());
174        }
175
176        match output.payload {
177            jsonrpsee_types::ResponsePayload::Success(success) => {
178                // Correlate the result to its request. The id check is only applied
179                // to success responses: JSON-RPC allows a null `id` on errors where
180                // the request id could not be determined, so an error is returned
181                // as-is rather than masked by an id mismatch.
182                if output.id != expected_id {
183                    return Err(format!(
184                        "JSON-RPC response id {:?} does not match request id {expected_id:?}",
185                        output.id
186                    )
187                    .into());
188                }
189
190                Ok(serde_json::from_value(success.into_owned())?)
191            }
192            jsonrpsee_types::ResponsePayload::Error(failure) => Err(failure.to_string().into()),
193        }
194    }
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200    use std::sync::{Arc, Mutex};
201
202    /// Proves that `RpcRequestClient` times out instead of hanging indefinitely
203    /// when a server accepts a TCP connection but never sends a response.
204    #[tokio::test]
205    async fn rpc_client_timeout_on_unresponsive_server() {
206        let listener =
207            std::net::TcpListener::bind("127.0.0.1:0").expect("should bind to localhost");
208        let addr = listener.local_addr().expect("should have a local address");
209
210        // Hold the accepted stream so the connection stays open until the test ends.
211        let held_stream = Arc::new(Mutex::new(None));
212        let held_stream_clone = held_stream.clone();
213
214        // Accept the connection but never respond.
215        let accept_thread = std::thread::spawn(move || {
216            let (stream, _peer_addr) = listener.accept().expect("should accept a connection");
217            *held_stream_clone.lock().expect("lock poisoned") = Some(stream);
218        });
219
220        let short_timeout = Duration::from_secs(2);
221        let client = RpcRequestClient::new_with_timeout(addr, short_timeout);
222
223        // Outer timeout is a safety net — should never fire.
224        let result = tokio::time::timeout(
225            Duration::from_secs(30),
226            client.text_from_call("getinfo", "[]"),
227        )
228        .await;
229
230        let inner_result =
231            result.expect("outer safety timeout should not fire; client timeout should fire first");
232
233        let err =
234            inner_result.expect_err("request to unresponsive server should fail with timeout");
235        assert!(err.is_timeout(), "error should be a timeout, got: {err}");
236
237        drop(held_stream);
238        accept_thread
239            .join()
240            .expect("accept thread should exit cleanly");
241    }
242
243    /// Regression test for #10687: the client validates the `jsonrpc` version and
244    /// the response `id`, rejecting a mismatched id or a missing version.
245    #[test]
246    fn validates_version_and_id() {
247        // A matching version and id deserializes the result.
248        let matching = r#"{"jsonrpc":"2.0","result":42,"id":7}"#;
249        let value: i64 = RpcRequestClient::json_result_from_response_text(matching, Id::Number(7))
250            .expect("a matching response should deserialize");
251        assert_eq!(value, 42);
252
253        // A mismatched id is rejected.
254        let mismatched_id = r#"{"jsonrpc":"2.0","result":42,"id":8}"#;
255        assert!(
256            RpcRequestClient::json_result_from_response_text::<i64>(mismatched_id, Id::Number(7))
257                .is_err(),
258            "a response with a mismatched id should be rejected",
259        );
260
261        // A missing jsonrpc version is rejected.
262        let missing_version = r#"{"result":42,"id":7}"#;
263        assert!(
264            RpcRequestClient::json_result_from_response_text::<i64>(missing_version, Id::Number(7))
265                .is_err(),
266            "a response missing the jsonrpc version should be rejected",
267        );
268    }
269
270    /// An error response is returned as an error and surfaces the server's message,
271    /// even when its `id` is null (which JSON-RPC permits on some errors) and so
272    /// does not match the request id. See #10687.
273    #[test]
274    fn error_response_is_returned_regardless_of_id() {
275        let error =
276            r#"{"jsonrpc":"2.0","error":{"code":-32601,"message":"method not found"},"id":null}"#;
277        let err = RpcRequestClient::json_result_from_response_text::<i64>(error, Id::Number(7))
278            .expect_err("an error response should be returned as an error");
279        assert!(
280            err.to_string().contains("method not found"),
281            "the error should surface the server's message, got: {err}",
282        );
283    }
284}
zebra_node_services/rpc_client.rs

zebra_node_services/
rpc_client.rs
