evm_state_diff/
rpc.rs

1use std::fmt;
2use std::str::FromStr;
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::time::Duration;
5
6use serde_json::{Value, json};
7
8use crate::primitives::B256;
9use crate::trace::{TraceError, TxTrace, parse_traces};
10
11#[derive(Debug, thiserror::Error)]
12pub enum RpcError {
13    #[error("request to {url} failed: {source}")]
14    Transport { url: String, source: Box<ureq::Error> },
15    #[error("JSON-RPC error {code}: {message}{hint}")]
16    Rpc { code: i64, message: String, hint: &'static str },
17    #[error("malformed JSON-RPC response: {0}")]
18    Malformed(String),
19    #[error(transparent)]
20    Trace(#[from] TraceError),
21}
22
23#[derive(Clone, Debug, PartialEq, Eq)]
24pub enum BlockId {
25    Tag(&'static str),
26    Number(u64),
27    Hash(B256),
28}
29
30impl FromStr for BlockId {
31    type Err = String;
32
33    fn from_str(s: &str) -> Result<Self, String> {
34        const TAGS: [&str; 5] = ["latest", "earliest", "pending", "safe", "finalized"];
35        if let Some(tag) = TAGS.iter().find(|t| **t == s) {
36            return Ok(BlockId::Tag(tag));
37        }
38        if let Some(hex) = s.strip_prefix("0x") {
39            if hex.len() == 64 {
40                return s.parse().map(BlockId::Hash).map_err(|e| format!("{e}"));
41            }
42            return u64::from_str_radix(hex, 16)
43                .map(BlockId::Number)
44                .map_err(|_| format!("invalid block `{s}`"));
45        }
46        s.parse().map(BlockId::Number).map_err(|_| {
47            format!(
48                "invalid block `{s}`: use a number, a 0x-prefixed number or hash, or one of {}",
49                TAGS.join(", ")
50            )
51        })
52    }
53}
54
55impl fmt::Display for BlockId {
56    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
57        match self {
58            BlockId::Tag(t) => f.write_str(t),
59            BlockId::Number(n) => write!(f, "{n}"),
60            BlockId::Hash(h) => write!(f, "{h}"),
61        }
62    }
63}
64
65#[derive(Clone, Copy, Debug, Default)]
66pub struct TracerOptions {
67    pub disable_code: bool,
68    pub disable_storage: bool,
69}
70
71impl TracerOptions {
72    pub fn to_json(self) -> Value {
73        json!({
74            "tracer": "prestateTracer",
75            "tracerConfig": {
76                "diffMode": true,
77                "disableCode": self.disable_code,
78                "disableStorage": self.disable_storage,
79            },
80        })
81    }
82}
83
84/// A minimal blocking JSON-RPC client for the `debug` tracing endpoints.
85pub struct RpcClient {
86    url: String,
87    agent: ureq::Agent,
88    headers: Vec<(String, String)>,
89    retries: u32,
90    backoff: Duration,
91    next_id: AtomicU64,
92}
93
94/// Configures an [`RpcClient`].
95#[derive(Clone, Debug)]
96pub struct RpcClientBuilder {
97    url: String,
98    timeout: Duration,
99    headers: Vec<(String, String)>,
100    retries: u32,
101    backoff: Duration,
102}
103
104impl RpcClientBuilder {
105    /// Timeout for each request, including retries individually.
106    pub fn timeout(mut self, timeout: Duration) -> Self {
107        self.timeout = timeout;
108        self
109    }
110
111    /// Adds a header to every request, e.g. an API key for a hosted node.
112    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
113        self.headers.push((name.into(), value.into()));
114        self
115    }
116
117    /// Retries after connection failures and HTTP 429/502/503/504, with
118    /// exponential backoff starting at `backoff`. JSON-RPC errors are never
119    /// retried.
120    pub fn retries(mut self, retries: u32, backoff: Duration) -> Self {
121        self.retries = retries;
122        self.backoff = backoff;
123        self
124    }
125
126    pub fn build(self) -> RpcClient {
127        let agent = ureq::Agent::config_builder()
128            .timeout_global(Some(self.timeout))
129            .http_status_as_error(false)
130            .build();
131        RpcClient {
132            url: self.url,
133            agent: agent.into(),
134            headers: self.headers,
135            retries: self.retries,
136            backoff: self.backoff,
137            next_id: AtomicU64::new(1),
138        }
139    }
140}
141
142const RETRY_STATUS: [u16; 4] = [429, 502, 503, 504];
143
144impl RpcClient {
145    /// A client without retries or extra headers.
146    pub fn new(url: &str, timeout: Duration) -> Self {
147        Self::builder(url).timeout(timeout).build()
148    }
149
150    pub fn builder(url: &str) -> RpcClientBuilder {
151        RpcClientBuilder {
152            url: url.to_owned(),
153            timeout: Duration::from_secs(60),
154            headers: Vec::new(),
155            retries: 0,
156            backoff: Duration::from_millis(250),
157        }
158    }
159
160    pub fn call(&self, method: &str, params: Value) -> Result<Value, RpcError> {
161        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
162        let body = json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params});
163        let mut attempt = 0;
164        let (status, text) = loop {
165            match self.send(&body) {
166                Ok((status, _)) if RETRY_STATUS.contains(&status) && attempt < self.retries => {}
167                Err(_) if attempt < self.retries => {}
168                other => break other?,
169            }
170            std::thread::sleep(self.backoff.saturating_mul(1 << attempt.min(16)));
171            attempt += 1;
172        };
173        let value: Value = serde_json::from_str(&text).map_err(|_| {
174            let snippet: String = text.chars().take(200).collect();
175            RpcError::Malformed(format!("HTTP {status}: {snippet}"))
176        })?;
177        if let Some(err) = value.get("error") {
178            let code = err.get("code").and_then(Value::as_i64).unwrap_or(0);
179            let message = err.get("message").and_then(Value::as_str).unwrap_or("").to_owned();
180            let hint = match code {
181                -32601 => {
182                    " (the node does not expose the debug namespace; enable it, e.g. geth --http.api eth,debug)"
183                }
184                -32602 => {
185                    " (the node rejected the tracer options; it may not support prestateTracer diffMode)"
186                }
187                _ => "",
188            };
189            return Err(RpcError::Rpc { code, message, hint });
190        }
191        value
192            .get("result")
193            .cloned()
194            .ok_or_else(|| RpcError::Malformed(format!("HTTP {status}: no result field")))
195    }
196
197    fn send(&self, body: &Value) -> Result<(u16, String), RpcError> {
198        let transport = |e: ureq::Error| RpcError::Transport { url: self.url.clone(), source: Box::new(e) };
199        let mut req = self.agent.post(&self.url);
200        for (name, value) in &self.headers {
201            req = req.header(name, value);
202        }
203        let mut resp = req.send_json(body).map_err(transport)?;
204        let status = resp.status().as_u16();
205        let text = resp.body_mut().read_to_string().map_err(transport)?;
206        Ok((status, text))
207    }
208
209    pub fn trace_transaction(&self, hash: &B256, opts: TracerOptions) -> Result<TxTrace, RpcError> {
210        let result = self.call("debug_traceTransaction", json!([hash.to_string(), opts.to_json()]))?;
211        let mut traces = parse_traces(result)?;
212        let mut trace = traces.pop().ok_or(TraceError::UnknownFormat)?;
213        trace.tx_hash = Some(*hash);
214        Ok(trace)
215    }
216
217    /// Traces every block in `from..=to`, in order, as one list of
218    /// transactions. Compose them for the net change over the range.
219    pub fn trace_range(&self, from: u64, to: u64, opts: TracerOptions) -> Result<Vec<TxTrace>, RpcError> {
220        let mut traces = Vec::new();
221        for n in from..=to {
222            traces.extend(self.trace_block(&BlockId::Number(n), opts)?);
223        }
224        Ok(traces)
225    }
226
227    pub fn trace_block(&self, block: &BlockId, opts: TracerOptions) -> Result<Vec<TxTrace>, RpcError> {
228        let (method, id) = match block {
229            BlockId::Hash(h) => ("debug_traceBlockByHash", json!(h.to_string())),
230            BlockId::Number(n) => ("debug_traceBlockByNumber", json!(format!("0x{n:x}"))),
231            BlockId::Tag(t) => ("debug_traceBlockByNumber", json!(t)),
232        };
233        let result = self.call(method, json!([id, opts.to_json()]))?;
234        Ok(parse_traces(result)?)
235    }
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241
242    #[test]
243    fn block_ids() {
244        assert_eq!("latest".parse(), Ok(BlockId::Tag("latest")));
245        assert_eq!("12345".parse(), Ok(BlockId::Number(12345)));
246        assert_eq!("0x10".parse(), Ok(BlockId::Number(16)));
247        let hash = format!("0x{}", "ab".repeat(32));
248        assert!(matches!(hash.parse::<BlockId>(), Ok(BlockId::Hash(_))));
249        assert!("newest".parse::<BlockId>().is_err());
250        assert!("0xzz".parse::<BlockId>().is_err());
251    }
252
253    #[test]
254    fn tracer_config() {
255        let v = TracerOptions { disable_code: true, disable_storage: false }.to_json();
256        assert_eq!(v["tracer"], "prestateTracer");
257        assert_eq!(v["tracerConfig"]["diffMode"], true);
258        assert_eq!(v["tracerConfig"]["disableCode"], true);
259    }
260
261    /// Serves the given raw HTTP responses in order, one per connection,
262    /// and returns the request heads it received.
263    fn serve(responses: Vec<&'static str>) -> (String, std::thread::JoinHandle<Vec<String>>) {
264        use std::io::{BufRead, BufReader, Read, Write};
265        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
266        let url = format!("http://{}", listener.local_addr().unwrap());
267        let handle = std::thread::spawn(move || {
268            let mut heads = Vec::new();
269            for response in responses {
270                let (stream, _) = listener.accept().unwrap();
271                let mut reader = BufReader::new(stream);
272                let (mut head, mut len) = (String::new(), 0);
273                loop {
274                    let mut line = String::new();
275                    reader.read_line(&mut line).unwrap();
276                    if let Some(v) = line.to_ascii_lowercase().strip_prefix("content-length:") {
277                        len = v.trim().parse().unwrap();
278                    }
279                    if line == "\r\n" {
280                        break;
281                    }
282                    head.push_str(&line);
283                }
284                reader.read_exact(&mut vec![0; len]).unwrap();
285                reader.get_mut().write_all(response.as_bytes()).unwrap();
286                heads.push(head);
287            }
288            heads
289        });
290        (url, handle)
291    }
292
293    fn http(status: &str, body: &str) -> &'static str {
294        format!("HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len())
295            .leak()
296    }
297
298    #[test]
299    fn retries_transient_http_errors() {
300        let ok = http("200 OK", r#"{"jsonrpc":"2.0","id":1,"result":"0x1"}"#);
301        let (url, server) =
302            serve(vec![http("503 Service Unavailable", ""), http("429 Too Many Requests", ""), ok]);
303        let client = RpcClient::builder(&url)
304            .timeout(Duration::from_secs(5))
305            .retries(2, Duration::from_millis(1))
306            .header("x-api-key", "secret")
307            .build();
308        assert_eq!(client.call("eth_chainId", json!([])).unwrap(), json!("0x1"));
309        let heads = server.join().unwrap();
310        assert_eq!(heads.len(), 3);
311        assert!(heads.iter().all(|h| h.to_ascii_lowercase().contains("x-api-key: secret")));
312    }
313
314    #[test]
315    fn gives_up_after_the_retry_budget() {
316        let (url, server) =
317            serve(vec![http("503 Service Unavailable", "busy"), http("503 Service Unavailable", "busy")]);
318        let client = RpcClient::builder(&url).retries(1, Duration::from_millis(1)).build();
319        assert!(
320            matches!(client.call("eth_chainId", json!([])), Err(RpcError::Malformed(m)) if m.contains("503"))
321        );
322        server.join().unwrap();
323    }
324
325    #[test]
326    fn json_rpc_errors_are_not_retried() {
327        let err = http("200 OK", r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"no debug"}}"#);
328        let (url, server) = serve(vec![err]);
329        let client = RpcClient::builder(&url).retries(3, Duration::from_millis(1)).build();
330        assert!(matches!(
331            client.call("debug_traceTransaction", json!([])),
332            Err(RpcError::Rpc { code: -32601, .. })
333        ));
334        assert_eq!(server.join().unwrap().len(), 1);
335    }
336
337    #[test]
338    fn unreachable_node_is_a_transport_error() {
339        let client = RpcClient::new("http://127.0.0.1:1", Duration::from_secs(2));
340        assert!(matches!(client.call("eth_chainId", json!([])), Err(RpcError::Transport { .. })));
341    }
342}
evm_state_diff/rpc.rs

evm_state_diff/
rpc.rs
