1use 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
20pub const BAL_HASH_FIELD: &str = "blockAccessListHash";
22pub const BAL_METHOD: &str = "eth_getBlockAccessList";
24pub const MAX_BODY_BYTES: u64 = 64 * 1024 * 1024;
26
27pub struct JsonRpcSource {
30 url: String,
31 client: reqwest::Client,
32}
33
34#[derive(Deserialize)]
35struct RpcResponse {
36 result: Option<Value>,
37 error: Option<RpcError>,
38}
39
40#[derive(Deserialize)]
41struct RpcError {
42 code: i64,
43 message: String,
44}
45
46impl JsonRpcSource {
47 pub fn new(url: impl Into<String>) -> Self {
51 let client = Self::hardened_client().unwrap_or_default();
55 Self {
56 url: url.into(),
57 client,
58 }
59 }
60
61 fn hardened_client() -> reqwest::Result<reqwest::Client> {
62 reqwest::Client::builder()
63 .timeout(std::time::Duration::from_secs(30))
64 .redirect(reqwest::redirect::Policy::none())
65 .build()
66 }
67
68 pub fn try_new(url: impl Into<String>) -> Result<Self> {
70 let client = Self::hardened_client()
71 .map_err(|e| SourceError::Transport(format!("http client: {e}")))?;
72 Ok(Self {
73 url: url.into(),
74 client,
75 })
76 }
77
78 pub async fn call(&self, method: &str, params: Value) -> Result<Value> {
80 let body = json!({"jsonrpc": "2.0", "id": 1, "method": method, "params": params});
81 let http = self
82 .client
83 .post(&self.url)
84 .json(&body)
85 .send()
86 .await
87 .map_err(|e| SourceError::Transport(e.to_string()))?;
88 let status = http.status();
90 if !status.is_success() {
91 return Err(SourceError::Transport(format!(
92 "HTTP {} from {} for {method}",
93 status.as_u16(),
94 self.url
95 )));
96 }
97 if let Some(len) = http.content_length() {
101 if len > MAX_BODY_BYTES {
102 return Err(SourceError::Transport(format!(
103 "{method}: response of {len} bytes exceeds the {MAX_BODY_BYTES}-byte limit"
104 )));
105 }
106 }
107 let mut http = http;
108 let mut body: Vec<u8> = Vec::new();
109 while let Some(chunk) = http
110 .chunk()
111 .await
112 .map_err(|e| SourceError::Transport(format!("{method}: {e}")))?
113 {
114 if body.len() + chunk.len() > MAX_BODY_BYTES as usize {
115 return Err(SourceError::Transport(format!(
116 "{method}: response exceeds the {MAX_BODY_BYTES}-byte limit"
117 )));
118 }
119 body.extend_from_slice(&chunk);
120 }
121 let resp: RpcResponse = serde_json::from_slice(&body)
122 .map_err(|e| SourceError::Malformed(format!("{method}: {e}")))?;
123 if let Some(e) = resp.error {
124 return Err(SourceError::Rpc {
125 code: e.code,
126 message: e.message,
127 });
128 }
129 Ok(resp.result.unwrap_or(Value::Null))
130 }
131
132 async fn raw_block(&self, tag: Value) -> Result<Value> {
133 let v = self
134 .call("eth_getBlockByNumber", json!([tag, false]))
135 .await?;
136 if v.is_null() {
137 return Err(SourceError::BlockNotFound(match &tag {
138 Value::String(s) => parse_hex_u64(s).unwrap_or(u64::MAX),
139 _ => u64::MAX,
140 }));
141 }
142 Ok(v)
143 }
144
145 pub async fn bal(&self, number: u64) -> Result<BlockAccessList> {
147 let v = self
148 .call(BAL_METHOD, json!([format!("{number:#x}")]))
149 .await?;
150 if v.is_null() {
151 return Err(SourceError::NoBal(number));
152 }
153 Ok(BlockAccessList::from_rpc_json(&v)?)
154 }
155
156 async fn probe_block(&self, tag: Value) -> BalProbe {
159 let v = match self.raw_block(tag).await {
160 Ok(v) => v,
161 Err(e) => return BalProbe::Error(e.to_string()),
162 };
163 let header = match parse_header(&v) {
164 Ok(h) => h,
165 Err(e) => return BalProbe::Error(e.to_string()),
166 };
167 let bal = match self.bal(header.number).await {
168 Ok(b) => b,
169 Err(SourceError::NoBal(n)) => return BalProbe::Missing(n),
170 Err(e) => return BalProbe::Error(e.to_string()),
171 };
172 let computed = bal.hash();
173 match header.block_access_list_hash {
174 None => BalProbe::NoHashInHeader {
175 block: header.number,
176 accounts: bal.accounts.len(),
177 },
178 Some(expected) if expected == computed => BalProbe::Verified {
179 block: header.number,
180 accounts: bal.accounts.len(),
181 hash: computed,
182 },
183 Some(expected) => BalProbe::Mismatch {
184 block: header.number,
185 computed,
186 expected,
187 },
188 }
189 }
190
191 pub async fn probe(&self, old_block_age: u64) -> Result<ProbeReport> {
194 let head_v = self.raw_block(json!("latest")).await?;
195 let head = parse_header(&head_v)?;
196 let old_number = head.number.saturating_sub(old_block_age).max(1);
197 let chain_id = self
198 .call("eth_chainId", json!([]))
199 .await
200 .ok()
201 .and_then(|v| v.as_str().and_then(|s| parse_hex_u64(s).ok()));
202 let client_version = self
203 .call("web3_clientVersion", json!([]))
204 .await
205 .ok()
206 .and_then(|v| v.as_str().map(String::from));
207 let head_fields = head_v
208 .as_object()
209 .map(|o| o.keys().cloned().collect())
210 .unwrap_or_default();
211 let head_probe = self.probe_block(json!(format!("{:#x}", head.number))).await;
212 let old_probe = self.probe_block(json!(format!("{old_number:#x}"))).await;
213 let earliest_probe = self.probe_block(json!("0x1")).await;
214 let proof_window = self.measure_proof_window(head.number).await;
215
216 Ok(ProbeReport {
217 client_version,
218 chain_id,
219 head: head.number,
220 head_fields,
221 head_probe,
222 old_probe,
223 earliest_probe,
224 proof_window,
225 })
226 }
227
228 pub async fn measure_proof_window(&self, head: u64) -> std::result::Result<u64, String> {
232 let mut ok: Option<u64> = None;
233 for k in [0u64, 1, 2, 4, 8, 16, 32, 64, 128, 256, 1024, 4096] {
234 if k > head {
235 break;
236 }
237 let r = self
238 .call(
239 "eth_getProof",
240 json!([
241 Address::ZERO,
242 Vec::<B256>::new(),
243 format!("{:#x}", head - k)
244 ]),
245 )
246 .await;
247 match r {
248 Ok(_) => ok = Some(k),
249 Err(e) => {
250 if ok.is_none() {
251 return Err(e.to_string());
252 }
253 break;
254 }
255 }
256 }
257 Ok(ok.unwrap_or(0))
258 }
259}
260
261#[derive(Debug, Clone)]
263pub enum BalProbe {
264 Verified {
266 block: u64,
268 accounts: usize,
270 hash: B256,
272 },
273 NoHashInHeader {
275 block: u64,
277 accounts: usize,
279 },
280 Mismatch {
282 block: u64,
284 computed: B256,
286 expected: B256,
288 },
289 Missing(u64),
291 Error(String),
293}
294
295#[derive(Debug, Clone)]
297pub struct ProbeReport {
298 pub client_version: Option<String>,
300 pub chain_id: Option<u64>,
302 pub head: u64,
304 pub head_fields: Vec<String>,
306 pub head_probe: BalProbe,
308 pub old_probe: BalProbe,
310 pub earliest_probe: BalProbe,
312 pub proof_window: std::result::Result<u64, String>,
315}
316
317fn expect_number(h: Header, requested: u64) -> Result<Header> {
320 if h.number != requested {
321 return Err(SourceError::Malformed(format!(
322 "asked for block {requested}, node answered with block {}",
323 h.number
324 )));
325 }
326 Ok(h)
327}
328
329fn parse_hex_u64(s: &str) -> Result<u64> {
330 let s = s.strip_prefix("0x").unwrap_or(s);
331 u64::from_str_radix(s, 16).map_err(|e| SourceError::Malformed(format!("u64 {s}: {e}")))
332}
333
334fn parse_b256(v: &Value, name: &str) -> Result<B256> {
335 let s = v
336 .as_str()
337 .ok_or_else(|| SourceError::Malformed(format!("{name}: not a string")))?;
338 s.parse::<B256>()
339 .map_err(|e| SourceError::Malformed(format!("{name}: {e}")))
340}
341
342fn field<'a>(v: &'a Value, name: &str) -> Result<&'a Value> {
343 v.get(name)
344 .ok_or_else(|| SourceError::Malformed(format!("missing field {name}")))
345}
346
347fn parse_header(v: &Value) -> Result<Header> {
348 let num = field(v, "number")?
349 .as_str()
350 .ok_or_else(|| SourceError::Malformed("number".into()))?;
351 let ts = field(v, "timestamp")?
352 .as_str()
353 .ok_or_else(|| SourceError::Malformed("timestamp".into()))?;
354 let bal_hash = v
355 .get(BAL_HASH_FIELD)
356 .filter(|x| !x.is_null())
357 .map(|x| parse_b256(x, BAL_HASH_FIELD))
358 .transpose()?;
359 Ok(Header {
360 number: parse_hex_u64(num)?,
361 hash: parse_b256(field(v, "hash")?, "hash")?,
362 parent_hash: parse_b256(field(v, "parentHash")?, "parentHash")?,
363 state_root: parse_b256(field(v, "stateRoot")?, "stateRoot")?,
364 timestamp: parse_hex_u64(ts)?,
365 block_access_list_hash: bal_hash,
366 })
367}
368
369#[async_trait]
370impl BalSource for JsonRpcSource {
371 async fn header(&self, number: u64) -> Result<Header> {
372 let v = self.raw_block(json!(format!("{number:#x}"))).await?;
373 expect_number(parse_header(&v)?, number)
374 }
375
376 async fn head(&self) -> Result<u64> {
377 let v = self.call("eth_blockNumber", json!([])).await?;
378 parse_hex_u64(
379 v.as_str()
380 .ok_or_else(|| SourceError::Malformed("blockNumber".into()))?,
381 )
382 }
383
384 async fn finalized(&self) -> Result<u64> {
385 let v = self.raw_block(json!("finalized")).await?;
386 Ok(parse_header(&v)?.number)
387 }
388
389 async fn block(&self, number: u64) -> Result<SourcedBlock> {
390 let v = self.raw_block(json!(format!("{number:#x}"))).await?;
391 let header = expect_number(parse_header(&v)?, number)?;
392 let bal = JsonRpcSource::bal(self, number).await?;
393 Ok(SourcedBlock { header, bal })
394 }
395
396 async fn bal(&self, number: u64) -> Result<BlockAccessList> {
397 JsonRpcSource::bal(self, number).await
398 }
399}
400
401#[derive(Deserialize)]
402#[serde(rename_all = "camelCase")]
403struct ProofResp {
404 balance: U256,
405 nonce: U256,
406 code_hash: B256,
407 storage_hash: B256,
408 account_proof: Vec<Bytes>,
409 storage_proof: Vec<StorageProofResp>,
410}
411
412#[derive(Deserialize)]
413struct StorageProofResp {
414 key: U256,
415 value: U256,
416 proof: Vec<Bytes>,
417}
418
419#[async_trait]
420impl StateSource for JsonRpcSource {
421 async fn proof(&self, addr: Address, slots: &[B256], block: u64) -> Result<AccountProof> {
422 let v = self
423 .call("eth_getProof", json!([addr, slots, format!("{block:#x}")]))
424 .await?;
425 let p: ProofResp =
426 serde_json::from_value(v).map_err(|e| SourceError::Malformed(format!("proof: {e}")))?;
427 let nonce: u64 = p
428 .nonce
429 .try_into()
430 .map_err(|_| SourceError::Malformed(format!("proof nonce {} exceeds u64", p.nonce)))?;
431 Ok(AccountProof {
432 address: addr,
433 balance: p.balance,
434 nonce,
435 code_hash: p.code_hash,
436 storage_hash: p.storage_hash,
437 account_proof: p.account_proof,
438 storage_proofs: p
439 .storage_proof
440 .into_iter()
441 .map(|s| StorageProof {
442 key: B256::from(s.key.to_be_bytes::<32>()),
443 value: s.value,
444 proof: s.proof,
445 })
446 .collect(),
447 })
448 }
449}