bal_source/
jsonrpc.rs

1//! JSON-RPC source. Day-0 findings on Platåberget (reth 2.5.0, 2026-08-29):
2//!
3//! - the block object carries `blockAccessListHash` but **not** the BAL body;
4//! - the body comes from `eth_getBlockAccessList(blockNumberOrTagOrHash)`,
5//!   decoded to JSON (execution-apis), available back to block 1;
6//! - `debug_getRawBlockAccessList` exists in reth but public gateways block
7//!   `debug_*`.
8//!
9//! Also hosts the day-0 probe.
10
11use crate::{
12    AccountProof, BalSource, Header, Result, SourceError, SourcedBlock, StateSource, StorageProof,
13};
14use alloy_primitives::{Address, Bytes, B256, U256};
15use async_trait::async_trait;
16use bal_codec::BlockAccessList;
17use serde::Deserialize;
18use serde_json::{json, Value};
19
20/// Header field carrying `keccak(rlp(bal))` on the block object.
21pub const BAL_HASH_FIELD: &str = "blockAccessListHash";
22/// JSON-RPC method serving the BAL body (execution-apis).
23pub const BAL_METHOD: &str = "eth_getBlockAccessList";
24/// Attempts per call on transport failure (gateways behind a CDN answer 5xx
25/// under load and drop bodies); RPC errors are never retried.
26const RETRIES: u32 = 4;
27
28/// Largest JSON-RPC response body accepted, in bytes.
29pub const MAX_BODY_BYTES: u64 = 64 * 1024 * 1024;
30
31/// [`BalSource`] + [`StateSource`] over plain JSON-RPC. One request per call;
32/// no batching, no retries — callers own that policy.
33pub struct JsonRpcSource {
34    url: String,
35    client: reqwest::Client,
36}
37
38#[derive(Deserialize)]
39struct RpcResponse {
40    result: Option<Value>,
41    error: Option<RpcError>,
42}
43
44#[derive(Deserialize)]
45struct RpcError {
46    code: i64,
47    message: String,
48}
49
50impl JsonRpcSource {
51    /// Talk to the JSON-RPC endpoint at `url`. Requests time out after 30 s
52    /// so a stalled gateway cannot hang a sync forever; redirects are not
53    /// followed, so a request never silently goes to a host you did not name.
54    pub fn new(url: impl Into<String>) -> Self {
55        // The builder only fails if the TLS backend cannot initialise; in that
56        // case no client would work, so a default one is no worse — but it
57        // must not silently drop the policies when they *can* be applied.
58        let client = Self::hardened_client().unwrap_or_default();
59        Self {
60            url: url.into(),
61            client,
62        }
63    }
64
65    fn hardened_client() -> reqwest::Result<reqwest::Client> {
66        reqwest::Client::builder()
67            .timeout(std::time::Duration::from_secs(30))
68            .redirect(reqwest::redirect::Policy::none())
69            .build()
70    }
71
72    /// Like [`JsonRpcSource::new`] but surfaces a client-construction failure.
73    pub fn try_new(url: impl Into<String>) -> Result<Self> {
74        let client = Self::hardened_client()
75            .map_err(|e| SourceError::Transport(format!("http client: {e}")))?;
76        Ok(Self {
77            url: url.into(),
78            client,
79        })
80    }
81
82    /// Raw JSON-RPC call; returns the `result` member or the error object as
83    /// [`SourceError::Rpc`]. Transport failures (5xx from a gateway, a
84    /// connection dropped mid-body) are retried a few times with backoff;
85    /// RPC errors and malformed JSON are not.
86    pub async fn call(&self, method: &str, params: Value) -> Result<Value> {
87        let mut delay = std::time::Duration::from_millis(400);
88        for attempt in 0..RETRIES {
89            match self.call_once(method, params.clone()).await {
90                Err(SourceError::Transport(e)) if attempt + 1 < RETRIES => {
91                    tracing::debug!(method, attempt, %e, "transport error; retrying");
92                    tokio::time::sleep(delay).await;
93                    delay *= 2;
94                }
95                other => return other,
96            }
97        }
98        unreachable!("the loop returns on its last attempt")
99    }
100
101    async fn call_once(&self, method: &str, params: Value) -> Result<Value> {
102        let body = json!({"jsonrpc": "2.0", "id": 1, "method": method, "params": params});
103        let http = self
104            .client
105            .post(&self.url)
106            .json(&body)
107            .send()
108            .await
109            .map_err(|e| SourceError::Transport(e.to_string()))?;
110        // Gateways answer 5xx with HTML; say so instead of "invalid JSON".
111        let status = http.status();
112        if !status.is_success() {
113            return Err(SourceError::Transport(format!(
114                "HTTP {} from {} for {method}",
115                status.as_u16(),
116                self.url
117            )));
118        }
119        // Read the body with a hard cap: a node must not be able to make us
120        // allocate without bound. The EIP caps a BAL at 8 MiB of RLP; its
121        // JSON form is a few times larger. 64 MiB is generous.
122        if let Some(len) = http.content_length() {
123            if len > MAX_BODY_BYTES {
124                return Err(SourceError::TooLarge {
125                    method: method.into(),
126                    bytes: len,
127                    limit: MAX_BODY_BYTES,
128                });
129            }
130        }
131        let mut http = http;
132        let mut body: Vec<u8> = Vec::new();
133        while let Some(chunk) = http
134            .chunk()
135            .await
136            .map_err(|e| SourceError::Transport(format!("{method}: {e}")))?
137        {
138            if body.len() + chunk.len() > MAX_BODY_BYTES as usize {
139                return Err(SourceError::TooLarge {
140                    method: method.into(),
141                    bytes: (body.len() + chunk.len()) as u64,
142                    limit: MAX_BODY_BYTES,
143                });
144            }
145            body.extend_from_slice(&chunk);
146        }
147        let resp: RpcResponse = serde_json::from_slice(&body)
148            .map_err(|e| SourceError::Malformed(format!("{method}: {e}")))?;
149        if let Some(e) = resp.error {
150            return Err(SourceError::Rpc {
151                code: e.code,
152                message: e.message,
153            });
154        }
155        Ok(resp.result.unwrap_or(Value::Null))
156    }
157
158    async fn raw_block(&self, tag: Value) -> Result<Value> {
159        let v = self
160            .call("eth_getBlockByNumber", json!([tag, false]))
161            .await?;
162        if v.is_null() {
163            return Err(SourceError::BlockNotFound(match &tag {
164                Value::String(s) => parse_hex_u64(s).unwrap_or(u64::MAX),
165                _ => u64::MAX,
166            }));
167        }
168        Ok(v)
169    }
170
171    /// Fetch and decode the BAL of `number` via `eth_getBlockAccessList`.
172    pub async fn bal(&self, number: u64) -> Result<BlockAccessList> {
173        let v = self
174            .call(BAL_METHOD, json!([format!("{number:#x}")]))
175            .await?;
176        if v.is_null() {
177            return Err(SourceError::NoBal(number));
178        }
179        Ok(BlockAccessList::from_rpc_json(&v)?)
180    }
181
182    /// Fetch header + BAL and check the BAL against the header. Used by the
183    /// probe; `sync` does the same check itself.
184    async fn probe_block(&self, tag: Value) -> BalProbe {
185        let v = match self.raw_block(tag).await {
186            Ok(v) => v,
187            Err(e) => return BalProbe::Error(e.to_string()),
188        };
189        let header = match parse_header(&v) {
190            Ok(h) => h,
191            Err(e) => return BalProbe::Error(e.to_string()),
192        };
193        let bal = match self.bal(header.number).await {
194            Ok(b) => b,
195            Err(SourceError::NoBal(n)) => return BalProbe::Missing(n),
196            Err(e) => return BalProbe::Error(e.to_string()),
197        };
198        let computed = bal.hash();
199        match header.block_access_list_hash {
200            None => BalProbe::NoHashInHeader {
201                block: header.number,
202                accounts: bal.accounts.len(),
203            },
204            Some(expected) if expected == computed => BalProbe::Verified {
205                block: header.number,
206                accounts: bal.accounts.len(),
207                hash: computed,
208            },
209            Some(expected) => BalProbe::Mismatch {
210                block: header.number,
211                computed,
212                expected,
213            },
214        }
215    }
216
217    /// Day-0 probe: Q1 (is the BAL served?), Q2 (for old blocks too?), and
218    /// does our codec reproduce the header hash on real data.
219    pub async fn probe(&self, old_block_age: u64) -> Result<ProbeReport> {
220        let head_v = self.raw_block(json!("latest")).await?;
221        let head = parse_header(&head_v)?;
222        let old_number = head.number.saturating_sub(old_block_age).max(1);
223        let chain_id = self
224            .call("eth_chainId", json!([]))
225            .await
226            .ok()
227            .and_then(|v| v.as_str().and_then(|s| parse_hex_u64(s).ok()));
228        let client_version = self
229            .call("web3_clientVersion", json!([]))
230            .await
231            .ok()
232            .and_then(|v| v.as_str().map(String::from));
233        let head_fields = head_v
234            .as_object()
235            .map(|o| o.keys().cloned().collect())
236            .unwrap_or_default();
237        let head_probe = self.probe_block(json!(format!("{:#x}", head.number))).await;
238        let old_probe = self.probe_block(json!(format!("{old_number:#x}"))).await;
239        let earliest_probe = self.probe_block(json!("0x1")).await;
240        let proof_window = self.measure_proof_window(head.number).await;
241
242        Ok(ProbeReport {
243            client_version,
244            chain_id,
245            head: head.number,
246            head_fields,
247            head_probe,
248            old_probe,
249            earliest_probe,
250            proof_window,
251        })
252    }
253
254    /// Largest distance behind head at which `eth_getProof` still answers.
255    /// reth: `--rpc.eth-proof-window` (default 0 = head only). `Err` if even
256    /// the head is refused (no bootstrap possible at all).
257    pub async fn measure_proof_window(&self, head: u64) -> std::result::Result<u64, String> {
258        let mut ok: Option<u64> = None;
259        for k in [0u64, 1, 2, 4, 8, 16, 32, 64, 128, 256, 1024, 4096] {
260            if k > head {
261                break;
262            }
263            let r = self
264                .call(
265                    "eth_getProof",
266                    json!([
267                        Address::ZERO,
268                        Vec::<B256>::new(),
269                        format!("{:#x}", head - k)
270                    ]),
271                )
272                .await;
273            match r {
274                Ok(_) => ok = Some(k),
275                Err(e) => {
276                    if ok.is_none() {
277                        return Err(e.to_string());
278                    }
279                    break;
280                }
281            }
282        }
283        Ok(ok.unwrap_or(0))
284    }
285}
286
287/// Outcome of fetching one block's BAL and checking it against its header.
288#[derive(Debug, Clone)]
289pub enum BalProbe {
290    /// BAL served and `keccak(rlp(bal)) == header.blockAccessListHash`.
291    Verified {
292        /// Block number.
293        block: u64,
294        /// Accounts in the BAL.
295        accounts: usize,
296        /// The matching hash.
297        hash: B256,
298    },
299    /// BAL served but the header has no hash field (pre-fork block or client gap).
300    NoHashInHeader {
301        /// Block number.
302        block: u64,
303        /// Accounts in the BAL.
304        accounts: usize,
305    },
306    /// BAL served, hash differs: codec/spec drift. Nothing should be built on this.
307    Mismatch {
308        /// Block number.
309        block: u64,
310        /// What this codec computed.
311        computed: B256,
312        /// What the header says.
313        expected: B256,
314    },
315    /// `eth_getBlockAccessList` returned null.
316    Missing(u64),
317    /// Transport or decoding failure.
318    Error(String),
319}
320
321/// Day-0 findings for one endpoint. Printed by `balq probe`.
322#[derive(Debug, Clone)]
323pub struct ProbeReport {
324    /// `web3_clientVersion`, if served.
325    pub client_version: Option<String>,
326    /// `eth_chainId`, if served.
327    pub chain_id: Option<u64>,
328    /// Head block number at probe time.
329    pub head: u64,
330    /// Field names on the head block object (to spot renamed BAL fields).
331    pub head_fields: Vec<String>,
332    /// Q1: the head block.
333    pub head_probe: BalProbe,
334    /// Q2: a block `age` blocks back.
335    pub old_probe: BalProbe,
336    /// Q2: block 1.
337    pub earliest_probe: BalProbe,
338    /// Measured `eth_getProof` window (blocks behind head still served), or
339    /// the error if proofs are not served at all.
340    pub proof_window: std::result::Result<u64, String>,
341}
342
343/// A node that answers block N with a header numbered M would otherwise get
344/// N's records filed under M. Refuse.
345fn expect_number(h: Header, requested: u64) -> Result<Header> {
346    if h.number != requested {
347        return Err(SourceError::Malformed(format!(
348            "asked for block {requested}, node answered with block {}",
349            h.number
350        )));
351    }
352    Ok(h)
353}
354
355fn parse_hex_u64(s: &str) -> Result<u64> {
356    let s = s.strip_prefix("0x").unwrap_or(s);
357    u64::from_str_radix(s, 16).map_err(|e| SourceError::Malformed(format!("u64 {s}: {e}")))
358}
359
360fn parse_b256(v: &Value, name: &str) -> Result<B256> {
361    let s = v
362        .as_str()
363        .ok_or_else(|| SourceError::Malformed(format!("{name}: not a string")))?;
364    s.parse::<B256>()
365        .map_err(|e| SourceError::Malformed(format!("{name}: {e}")))
366}
367
368fn field<'a>(v: &'a Value, name: &str) -> Result<&'a Value> {
369    v.get(name)
370        .ok_or_else(|| SourceError::Malformed(format!("missing field {name}")))
371}
372
373fn parse_header(v: &Value) -> Result<Header> {
374    let num = field(v, "number")?
375        .as_str()
376        .ok_or_else(|| SourceError::Malformed("number".into()))?;
377    let ts = field(v, "timestamp")?
378        .as_str()
379        .ok_or_else(|| SourceError::Malformed("timestamp".into()))?;
380    let bal_hash = v
381        .get(BAL_HASH_FIELD)
382        .filter(|x| !x.is_null())
383        .map(|x| parse_b256(x, BAL_HASH_FIELD))
384        .transpose()?;
385    let hash = parse_b256(field(v, "hash")?, "hash")?;
386    // The hash is recomputed from the header fields (EIP-7928 fields
387    // included): a node cannot hand out a header whose fields do not match
388    // the hash the chain is linked by.
389    let consensus: alloy_consensus::Header = serde_json::from_value(v.clone())
390        .map_err(|e| SourceError::Malformed(format!("header fields: {e}")))?;
391    let computed = consensus.hash_slow();
392    if computed != hash {
393        return Err(SourceError::Malformed(format!(
394            "header {num}: hash {hash} does not match its fields (keccak(rlp(header)) = {computed})"
395        )));
396    }
397    Ok(Header {
398        number: parse_hex_u64(num)?,
399        hash,
400        parent_hash: parse_b256(field(v, "parentHash")?, "parentHash")?,
401        state_root: parse_b256(field(v, "stateRoot")?, "stateRoot")?,
402        timestamp: parse_hex_u64(ts)?,
403        block_access_list_hash: bal_hash,
404    })
405}
406
407#[async_trait]
408impl BalSource for JsonRpcSource {
409    async fn header(&self, number: u64) -> Result<Header> {
410        let v = self.raw_block(json!(format!("{number:#x}"))).await?;
411        expect_number(parse_header(&v)?, number)
412    }
413
414    async fn head(&self) -> Result<u64> {
415        let v = self.call("eth_blockNumber", json!([])).await?;
416        parse_hex_u64(
417            v.as_str()
418                .ok_or_else(|| SourceError::Malformed("blockNumber".into()))?,
419        )
420    }
421
422    async fn finalized(&self) -> Result<u64> {
423        let v = self.raw_block(json!("finalized")).await?;
424        Ok(parse_header(&v)?.number)
425    }
426
427    async fn block(&self, number: u64) -> Result<SourcedBlock> {
428        let v = self.raw_block(json!(format!("{number:#x}"))).await?;
429        let header = expect_number(parse_header(&v)?, number)?;
430        let bal = JsonRpcSource::bal(self, number).await?;
431        Ok(SourcedBlock { header, bal })
432    }
433
434    async fn bal(&self, number: u64) -> Result<BlockAccessList> {
435        JsonRpcSource::bal(self, number).await
436    }
437}
438
439#[derive(Deserialize)]
440#[serde(rename_all = "camelCase")]
441struct ProofResp {
442    balance: U256,
443    nonce: U256,
444    code_hash: B256,
445    storage_hash: B256,
446    account_proof: Vec<Bytes>,
447    storage_proof: Vec<StorageProofResp>,
448}
449
450#[derive(Deserialize)]
451struct StorageProofResp {
452    key: U256,
453    value: U256,
454    proof: Vec<Bytes>,
455}
456
457#[async_trait]
458impl StateSource for JsonRpcSource {
459    async fn proof(&self, addr: Address, slots: &[B256], block: u64) -> Result<AccountProof> {
460        let v = self
461            .call("eth_getProof", json!([addr, slots, format!("{block:#x}")]))
462            .await?;
463        let p: ProofResp =
464            serde_json::from_value(v).map_err(|e| SourceError::Malformed(format!("proof: {e}")))?;
465        let nonce: u64 = p
466            .nonce
467            .try_into()
468            .map_err(|_| SourceError::Malformed(format!("proof nonce {} exceeds u64", p.nonce)))?;
469        Ok(AccountProof {
470            address: addr,
471            balance: p.balance,
472            nonce,
473            code_hash: p.code_hash,
474            storage_hash: p.storage_hash,
475            account_proof: p.account_proof,
476            storage_proofs: p
477                .storage_proof
478                .into_iter()
479                .map(|s| StorageProof {
480                    key: B256::from(s.key.to_be_bytes::<32>()),
481                    value: s.value,
482                    proof: s.proof,
483                })
484                .collect(),
485        })
486    }
487}
488
489#[cfg(test)]
490mod retry_tests {
491    #![allow(clippy::unwrap_used)]
492    use super::*;
493    use std::io::{Read, Write};
494    use std::net::TcpListener;
495    use std::sync::atomic::{AtomicUsize, Ordering};
496    use std::sync::Arc;
497
498    /// A one-thread HTTP server that answers each connection with the next
499    /// canned response and counts how many requests it saw.
500    fn serve(responses: Vec<String>) -> (String, Arc<AtomicUsize>) {
501        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
502        let url = format!("http://{}", listener.local_addr().unwrap());
503        let seen = Arc::new(AtomicUsize::new(0));
504        let counter = seen.clone();
505        std::thread::spawn(move || {
506            for resp in responses {
507                let (mut s, _) = listener.accept().unwrap();
508                let mut buf = [0u8; 4096];
509                let _ = s.read(&mut buf);
510                counter.fetch_add(1, Ordering::SeqCst);
511                let _ = s.write_all(resp.as_bytes());
512            }
513        });
514        (url, seen)
515    }
516
517    fn http(status: &str, body: &str) -> String {
518        format!(
519            "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
520            body.len()
521        )
522    }
523
524    #[tokio::test]
525    async fn transport_failure_is_retried_then_succeeds() {
526        let ok = r#"{"jsonrpc":"2.0","id":1,"result":"0x10"}"#;
527        let (url, seen) = serve(vec![
528            http("502 Bad Gateway", "<html>cloudflare</html>"),
529            http("503 Service Unavailable", ""),
530            http("200 OK", ok),
531        ]);
532        let src = JsonRpcSource::new(&url);
533        let v = src.call("eth_blockNumber", json!([])).await.unwrap();
534        assert_eq!(v, json!("0x10"));
535        assert_eq!(seen.load(Ordering::SeqCst), 3);
536    }
537
538    #[tokio::test]
539    async fn rpc_errors_and_oversized_bodies_are_not_retried() {
540        let rpc_err = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32602,"message":"nope"}}"#;
541        let (url, seen) = serve(vec![http("200 OK", rpc_err), http("200 OK", rpc_err)]);
542        let src = JsonRpcSource::new(&url);
543        let e = src.call("eth_getProof", json!([])).await.unwrap_err();
544        assert!(matches!(e, SourceError::Rpc { code: -32602, .. }), "{e}");
545        assert_eq!(seen.load(Ordering::SeqCst), 1, "an RPC error is final");
546
547        let huge = format!(
548            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
549            MAX_BODY_BYTES + 1
550        );
551        let (url, seen) = serve(vec![huge.clone(), huge]);
552        let src = JsonRpcSource::new(&url);
553        let e = src
554            .call("eth_getBlockAccessList", json!([]))
555            .await
556            .unwrap_err();
557        assert!(matches!(e, SourceError::TooLarge { .. }), "{e}");
558        assert_eq!(seen.load(Ordering::SeqCst), 1, "the cap is final too");
559    }
560}
bal_source/jsonrpc.rs

bal_source/
jsonrpc.rs
