From b4bdf9050225a1810dbfaacfe9dfd14ec110e8e0 Mon Sep 17 00:00:00 2001 From: Francis Li Date: Mon, 5 Oct 2026 19:09:45 +0000 Subject: [PATCH] Read historical Beacon blobs filtered by their actual content Amp-Thread-ID: https://ampcode.com/threads/T-01a10d69-00b6-7378-855f-af2b1a148692 Co-authored-by: Amp --- crates/anvil/src/eth/backend/beacon.rs | 150 +++++++++++++++++++++++-- 1 file changed, 141 insertions(+), 9 deletions(-) diff --git a/crates/anvil/src/eth/backend/beacon.rs b/crates/anvil/src/eth/backend/beacon.rs index d5dcc7d32..7888ca969 100644 --- a/crates/anvil/src/eth/backend/beacon.rs +++ b/crates/anvil/src/eth/backend/beacon.rs @@ -1,20 +1,41 @@ -//! The immutable Beacon slot schedule of an execution fork. +//! Historical Beacon reads and the immutable slot schedule of an execution fork. -use alloy_rpc_types_beacon::genesis::{GenesisData, GenesisResponse}; +use alloy_consensus::{Blob, EnvKzgSettings}; +use alloy_eips::eip4844::kzg_to_versioned_hash; +use alloy_primitives::B256; +use alloy_rpc_types_beacon::{ + genesis::{GenesisData, GenesisResponse}, + sidecar::GetBlobsResponse, +}; use eyre::{Result, ensure, eyre}; use reqwest::{Client, StatusCode, Url, redirect::Policy}; use serde::de::DeserializeOwned; -use std::time::Duration; +use std::{fmt, time::Duration}; +// JSON encodes each 128 KiB blob as hex. This permits over 100 blobs without accepting +// unbounded responses from an upstream provider. +const MAX_BLOB_RESPONSE_BYTES: usize = 32 * 1024 * 1024; const MAX_METADATA_RESPONSE_BYTES: usize = 1024 * 1024; -/// Beacon metadata belonging to an execution fork. -#[derive(Clone, Debug)] +/// Beacon connectivity and metadata belonging to an execution fork. +#[derive(Clone)] pub struct ForkBeacon { + url: Url, + client: Client, pub genesis: GenesisData, pub seconds_per_slot: u64, } +impl fmt::Debug for ForkBeacon { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // Endpoint paths and query parameters may contain credentials. + f.debug_struct("ForkBeacon") + .field("genesis", &self.genesis) + .field("seconds_per_slot", &self.seconds_per_slot) + .finish_non_exhaustive() + } +} + impl ForkBeacon { /// Captures Beacon metadata before the fork starts serving requests. pub async fn connect(endpoint: &str, timeout: Duration, fork_timestamp: u64) -> Result { @@ -25,6 +46,7 @@ impl ForkBeacon { &client, &url, "eth/v1/beacon/genesis", + &[], MAX_METADATA_RESPONSE_BYTES, ) .await? @@ -34,6 +56,7 @@ impl ForkBeacon { &client, &url, "eth/v1/config/spec", + &[], MAX_METADATA_RESPONSE_BYTES, ) .await? @@ -50,7 +73,7 @@ impl ForkBeacon { elapsed.is_multiple_of(seconds_per_slot), "execution fork timestamp is not on the Beacon slot grid" ); - Ok(Self { genesis, seconds_per_slot }) + Ok(Self { url, client, genesis, seconds_per_slot }) } /// Converts a slot to its exact execution timestamp without wrapping on user input. @@ -58,10 +81,45 @@ impl ForkBeacon { slot.checked_mul(self.seconds_per_slot)?.checked_add(self.genesis.genesis_time) } + /// Reads historical blobs. The caller must enforce the inclusive execution fork boundary. + pub async fn blobs(&self, slot: u64, hashes: &[B256]) -> Result>> { + let Some(response) = Self::get::( + &self.client, + &self.url, + &format!("eth/v1/beacon/blobs/{slot}"), + hashes, + MAX_BLOB_RESPONSE_BYTES, + ) + .await? + else { + return Ok(None); + }; + let mut blobs = response.data; + if !hashes.is_empty() { + // Providers may ignore the filter. Apply it to the actual blob content, not + // to untrusted upstream labels; Base also validates the requested hashes. + let hashes = hashes.to_vec(); + blobs = tokio::task::spawn_blocking(move || -> Result> { + let settings = EnvKzgSettings::Default; + let mut filtered = Vec::new(); + for blob in blobs { + let commitment = settings.get().blob_to_kzg_commitment(&blob.0.into())?; + if hashes.contains(&kzg_to_versioned_hash(commitment.as_slice())) { + filtered.push(blob); + } + } + Ok(filtered) + }) + .await??; + } + Ok(Some(blobs)) + } + async fn get( client: &Client, endpoint: &Url, path: &str, + hashes: &[B256], max_bytes: usize, ) -> Result> { // Append to the endpoint's path and keep its query; either may carry credentials. @@ -70,7 +128,11 @@ impl ForkBeacon { .map_err(|_| eyre!("invalid fork Beacon URL"))? .pop_if_empty() .extend(path.split('/')); - let request = client.get(url).header("Accept", "application/json"); + let mut request = client.get(url).header("Accept", "application/json"); + if !hashes.is_empty() { + let hashes = hashes.iter().map(ToString::to_string).collect::>().join(","); + request = request.query(&[("versioned_hashes", hashes)]); + } let response = request.send().await.map_err(reqwest::Error::without_url)?; if response.status() == StatusCode::NOT_FOUND { return Ok(None); @@ -89,13 +151,15 @@ impl ForkBeacon { #[cfg(test)] mod tests { use super::*; - use alloy_primitives::{B256, FixedBytes}; + use alloy_consensus::{BlobTransactionSidecar, SidecarBuilder, SimpleCoder}; + use alloy_primitives::FixedBytes; use axum::{ Json, Router, - extract::RawQuery, + extract::{Path, RawQuery}, response::{IntoResponse, Redirect}, routing::get, }; + use std::sync::{Arc, Mutex}; use tokio::net::TcpListener; const TIMEOUT: Duration = Duration::from_secs(1); @@ -227,4 +291,72 @@ mod tests { .await; assert!(connect_err(&slow, FORK).await.contains("timed out")); } + + fn sidecar(data: &[u8]) -> BlobTransactionSidecar { + SidecarBuilder::::from_slice(data).build().unwrap() + } + + /// Connects through a credential-bearing endpoint whose slot 20 serves `blobs` regardless of + /// filters. Slot 16 is oversized, slot 18 fails, other slots are missing. Records blob queries. + async fn blob_upstream(blobs: Vec) -> (ForkBeacon, Arc>>) { + let queries = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&queries); + let handler = move |Path(slot): Path, RawQuery(query): RawQuery| { + recorded.lock().unwrap().push(query.unwrap_or_default()); + let data = blobs.clone(); + async move { + match slot { + 16 => vec![b' '; MAX_BLOB_RESPONSE_BYTES + 1].into_response(), + 18 => StatusCode::INTERNAL_SERVER_ERROR.into_response(), + 20 => Json(GetBlobsResponse { + execution_optimistic: false, + finalized: true, + data, + }) + .into_response(), + _ => StatusCode::NOT_FOUND.into_response(), + } + } + }; + let router = metadata("/secret-key", "12") + .route("/secret-key/eth/v1/beacon/blobs/{slot}", get(handler)); + let url = serve(router).await; + let beacon = ForkBeacon::connect(&format!("{url}/secret-key?token=secret"), TIMEOUT, FORK) + .await + .unwrap(); + (beacon, queries) + } + + #[tokio::test] + async fn blobs_are_filtered_by_actual_content() { + let [first, second, absent] = [b"first".as_slice(), b"second", b"absent"].map(sidecar); + let hash = |sidecar: &BlobTransactionSidecar| sidecar.versioned_hashes().next().unwrap(); + let (beacon, queries) = blob_upstream(vec![first.blobs[0], second.blobs[0]]).await; + assert!(!format!("{beacon:?}").contains("secret")); + + let all = beacon.blobs(20, &[]).await.unwrap().unwrap(); + assert!(all == [first.blobs[0], second.blobs[0]], "unfiltered blobs"); + let filtered = beacon.blobs(20, &[hash(&second)]).await.unwrap().unwrap(); + assert!(filtered == [second.blobs[0]], "filtered blobs"); + assert!(beacon.blobs(20, &[hash(&absent)]).await.unwrap().unwrap().is_empty()); + assert_eq!( + *queries.lock().unwrap(), + [ + "token=secret".to_string(), + format!("token=secret&versioned_hashes={}", hash(&second)), + format!("token=secret&versioned_hashes={}", hash(&absent)), + ] + ); + } + + #[tokio::test] + async fn blobs_distinguish_missing_from_upstream_failures() { + let (beacon, _) = blob_upstream(Vec::new()).await; + assert!(beacon.blobs(19, &[]).await.unwrap().is_none()); + assert!(beacon.blobs(20, &[]).await.unwrap().unwrap().is_empty()); + for (slot, error) in [(16, "exceeds size limit"), (18, "500")] { + let err = format!("{:?}", beacon.blobs(slot, &[]).await.unwrap_err()); + assert!(err.contains(error) && !err.contains("secret"), "slot {slot}: {err}"); + } + } }