From 85d41388736cd6e0135669de9730194d6da91013 Mon Sep 17 00:00:00 2001 From: Vaclav Barta Date: Wed, 20 Mar 2024 09:04:11 +0100 Subject: [PATCH] encapsulated BlobHttpClient --- .../src/blob_http_client.rs | 32 +++++++++++++++++ state-reconstruct-fetcher/src/l1_fetcher.rs | 36 ++++--------------- state-reconstruct-fetcher/src/lib.rs | 1 + state-reconstruct-fetcher/src/types/mod.rs | 11 +++--- state-reconstruct-fetcher/src/types/v3.rs | 27 ++++++-------- 5 files changed, 56 insertions(+), 51 deletions(-) create mode 100644 state-reconstruct-fetcher/src/blob_http_client.rs diff --git a/state-reconstruct-fetcher/src/blob_http_client.rs b/state-reconstruct-fetcher/src/blob_http_client.rs new file mode 100644 index 0000000..90a7a87 --- /dev/null +++ b/state-reconstruct-fetcher/src/blob_http_client.rs @@ -0,0 +1,32 @@ +use eyre::Result; + +pub struct BlobHttpClient { + client: reqwest::Client, + url_base: String, +} + +impl BlobHttpClient { + pub fn new(blob_url: String) -> Result { + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert( + "Accept", + reqwest::header::HeaderValue::from_static("application/json"), + ); + let client = reqwest::Client::builder() + .default_headers(headers) + .build()?; + Ok(Self { + client, + url_base: blob_url, + }) + } + + pub fn format_url(&self, kzg_commitment: &[u8]) -> String { + format!("{}0x{}", self.url_base, hex::encode(kzg_commitment)) + } + + pub async fn retrieve_url(&self, url: &str) -> Result { + let result = self.client.get(url).send().await?; + Ok(result) + } +} diff --git a/state-reconstruct-fetcher/src/l1_fetcher.rs b/state-reconstruct-fetcher/src/l1_fetcher.rs index 438d680..782807f 100644 --- a/state-reconstruct-fetcher/src/l1_fetcher.rs +++ b/state-reconstruct-fetcher/src/l1_fetcher.rs @@ -14,6 +14,7 @@ use tokio::{ use tokio_util::sync::CancellationToken; use crate::{ + blob_http_client::BlobHttpClient, constants::ethereum::{BLOB_BLOCK, BLOCK_STEP, BOOJUM_BLOCK, GENESIS_BLOCK, ZK_SYNC_ADDR}, database::InnerDB, metrics::L1Metrics, @@ -449,8 +450,7 @@ impl L1Fetcher { ) -> Result>> { let metrics = self.metrics.clone(); let contracts = self.contracts.clone(); - let client = make_client()?; - let blobs_url = self.config.blobs_url.clone(); + let client = BlobHttpClient::new(self.config.blobs_url.clone())?; Ok(tokio::spawn({ async move { let mut boojum_mode = false; @@ -479,15 +479,7 @@ impl L1Fetcher { } let blocks = loop { - match parse_calldata( - block_number, - &function, - &tx.input, - &client, - &blobs_url, - ) - .await - { + match parse_calldata(block_number, &function, &tx.input, &client).await { Ok(blks) => break blks, Err(e) => match e { ParseError::BlobStorageError(_) => { @@ -549,24 +541,11 @@ impl L1Fetcher { } } -fn make_client() -> Result { - let mut headers = reqwest::header::HeaderMap::new(); - headers.insert( - "Accept", - reqwest::header::HeaderValue::from_static("application/json"), - ); - let client = reqwest::Client::builder() - .default_headers(headers) - .build()?; - Ok(client) -} - pub async fn parse_calldata( l1_block_number: u64, commit_blocks_fn: &Function, calldata: &[u8], - client: &reqwest::Client, - blobs_url: &str, + client: &BlobHttpClient, ) -> Result, ParseError> { let mut parsed_input = commit_blocks_fn .decode_input(&calldata[4..]) @@ -606,7 +585,7 @@ pub async fn parse_calldata( // Parse blocks using [`CommitBlockInfoV1`] or [`CommitBlockInfoV2`] let mut block_infos = - parse_commit_block_info(&new_blocks_data, l1_block_number, client, blobs_url).await?; + parse_commit_block_info(&new_blocks_data, l1_block_number, client).await?; // Supplement every `CommitBlock` element with L1 block number information. block_infos .iter_mut() @@ -617,8 +596,7 @@ pub async fn parse_calldata( async fn parse_commit_block_info( data: &abi::Token, l1_block_number: u64, - client: &reqwest::Client, - blobs_url: &str, + client: &BlobHttpClient, ) -> Result, ParseError> { let abi::Token::Array(data) = data else { return Err(ParseError::InvalidCommitBlockInfo( @@ -630,7 +608,7 @@ async fn parse_commit_block_info( for d in data { let commit_block = { if l1_block_number >= BLOB_BLOCK { - CommitBlock::try_from_token_resolve(d, client, blobs_url).await? + CommitBlock::try_from_token_resolve(d, client).await? } else if l1_block_number >= BOOJUM_BLOCK { CommitBlock::try_from_token::(d)? } else { diff --git a/state-reconstruct-fetcher/src/lib.rs b/state-reconstruct-fetcher/src/lib.rs index 3f6de0b..e116019 100644 --- a/state-reconstruct-fetcher/src/lib.rs +++ b/state-reconstruct-fetcher/src/lib.rs @@ -1,5 +1,6 @@ #![feature(array_chunks)] #![feature(iter_next_chunk)] +pub mod blob_http_client; pub mod constants; pub mod database; pub mod l1_fetcher; diff --git a/state-reconstruct-fetcher/src/types/mod.rs b/state-reconstruct-fetcher/src/types/mod.rs index 618f4ac..95534cb 100644 --- a/state-reconstruct-fetcher/src/types/mod.rs +++ b/state-reconstruct-fetcher/src/types/mod.rs @@ -5,6 +5,7 @@ use serde_json_any_key::any_key_map; use thiserror::Error; use self::{v1::V1, v2::V2, v3::V3}; +use crate::blob_http_client::BlobHttpClient; // NOTE: We should probably make these more human-readable. pub mod common; @@ -104,11 +105,10 @@ impl CommitBlock { pub async fn try_from_token_resolve<'a>( value: &'a abi::Token, - client: &reqwest::Client, - blobs_url: &str, + client: &BlobHttpClient, ) -> Result { let commit_block_info = V3::try_from(value)?; - Self::from_commit_block_resolve(commit_block_info, client, blobs_url).await + Self::from_commit_block_resolve(commit_block_info, client).await } pub fn from_commit_block(block_type: CommitBlockInfo) -> Self { @@ -170,10 +170,9 @@ impl CommitBlock { pub async fn from_commit_block_resolve( block: V3, - client: &reqwest::Client, - blobs_url: &str, + client: &BlobHttpClient, ) -> Result { - let total_l2_to_l1_pubdata = block.parse_pubdata(client, blobs_url).await?; + let total_l2_to_l1_pubdata = block.parse_pubdata(client).await?; let mut initial_storage_changes = IndexMap::new(); let mut repeated_storage_changes = IndexMap::new(); let mut factory_deps = Vec::new(); diff --git a/state-reconstruct-fetcher/src/types/v3.rs b/state-reconstruct-fetcher/src/types/v3.rs index 19b1283..ae2b01d 100644 --- a/state-reconstruct-fetcher/src/types/v3.rs +++ b/state-reconstruct-fetcher/src/types/v3.rs @@ -11,7 +11,10 @@ use super::{ common::{parse_compressed_state_diffs, read_next_n_bytes, ExtractedToken}, L2ToL1Pubdata, ParseError, }; -use crate::constants::zksync::{L2_TO_L1_LOG_SERIALIZE_SIZE, PUBDATA_COMMITMENT_SIZE}; +use crate::{ + blob_http_client::BlobHttpClient, + constants::zksync::{L2_TO_L1_LOG_SERIALIZE_SIZE, PUBDATA_COMMITMENT_SIZE}, +}; /// `MAX_RETRIES` is the maximum number of retries on failed blob retrieval. const MAX_RETRIES: u8 = 5; @@ -100,16 +103,13 @@ impl TryFrom<&abi::Token> for V3 { impl V3 { pub async fn parse_pubdata( &self, - client: &reqwest::Client, - blobs_url: &str, + client: &BlobHttpClient, ) -> Result, ParseError> { let mut pointer = 0; let bytes = &self.pubdata_commitments[..]; match self.pubdata_source { PubdataSource::Calldata => parse_pubdata_from_calldata(bytes, &mut pointer, true), - PubdataSource::Blob => { - parse_pubdata_from_blobs(bytes, &mut pointer, client, blobs_url).await - } + PubdataSource::Blob => parse_pubdata_from_blobs(bytes, &mut pointer, client).await, } } } @@ -163,14 +163,13 @@ fn parse_pubdata_from_calldata( async fn parse_pubdata_from_blobs( bytes: &[u8], pointer: &mut usize, - client: &reqwest::Client, - blobs_url: &str, + client: &BlobHttpClient, ) -> Result, ParseError> { let mut l = bytes.len() - *pointer; let mut blobs = Vec::new(); while *pointer < l { let pubdata_commitment = &bytes[*pointer..*pointer + PUBDATA_COMMITMENT_SIZE]; - let blob = get_blob(&pubdata_commitment[48..96], client, blobs_url).await?; + let blob = get_blob(&pubdata_commitment[48..96], client).await?; let mut blob_bytes = ethereum_4844_data_into_zksync_pubdata(&blob); blobs.append(&mut blob_bytes); *pointer += PUBDATA_COMMITMENT_SIZE; @@ -186,14 +185,10 @@ async fn parse_pubdata_from_blobs( parse_pubdata_from_calldata(blobs_view, &mut pointer, false) } -async fn get_blob( - kzg_commitment: &[u8], - client: &reqwest::Client, - blobs_url: &str, -) -> Result, ParseError> { - let url = format!("{}0x{}", blobs_url, hex::encode(kzg_commitment)); +async fn get_blob(kzg_commitment: &[u8], client: &BlobHttpClient) -> Result, ParseError> { + let url = client.format_url(kzg_commitment); for attempt in 1..=MAX_RETRIES { - match client.get(url.clone()).send().await { + match client.retrieve_url(&url).await { Ok(response) => match response.text().await { Ok(text) => match get_blob_data(&text) { Ok(data) => {