migrate: ethers to alloy

Cargo.toml

@@ -19,6 +19,15 @@ cryo_cli = { version = "0.3.2", path = "./crates/cli" }
 cryo_freeze = { version = "0.3.2", path = "./crates/freeze" }
 cryo_to_df = { version = "0.3.2", path = "./crates/to_df" }
 
+alloy = { version = "0.1", features = [
+    "full",
+    "rpc-types-trace",
+    "provider-ws",
+    "provider-ipc",
+    "provider-debug-api",
+    "provider-trace-api",
+    "transport-ipc-mock",
+] }
 anstyle = "1.0.4"
 async-trait = "0.1.74"
 chrono = { version = "0.4.31", features = ["serde"] }
@@ -29,8 +38,6 @@ clap_cryo = { version = "4.3.21-cryo", features = [
 ] }
 colored = "2.0.4"
 color-print = "0.3.5"
-ethers = { version = "2.0.10", features = ["rustls", "ws", "ipc"] }
-ethers-core = "2.0.10"
 eyre = "0.6.8"
 futures = "0.3.29"
 governor = "0.6.0"

crates/cli/Cargo.toml

@@ -18,12 +18,12 @@ path = "src/main.rs"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+alloy = { workspace = true }
 anstyle = { workspace = true }
 clap_cryo = { workspace = true }
 color-print = { workspace = true }
 colored = { workspace = true }
 cryo_freeze = { workspace = true }
-ethers = { workspace = true }
 eyre = { workspace = true }
 governor = { workspace = true }
 hex = { workspace = true }

crates/cli/src/parse/blocks.rs

@@ -168,6 +168,7 @@ async fn parse_block_inputs(
     }
 }
 
+#[derive(Clone, Debug)]
 enum RangePosition {
     First,
     Last,
@@ -322,12 +323,12 @@ async fn parse_block_number(
     source: Arc<Source>,
 ) -> Result<u64, ParseError> {
     match (block_ref, range_position) {
-        ("latest", _) => source.get_block_number().await.map(|n| n.as_u64()).map_err(|_e| {
+        ("latest", _) => source.get_block_number().await.map_err(|_e| {
             ParseError::ParseError("Error retrieving latest block number".to_string())
         }),
         ("", RangePosition::First) => Ok(0),
         ("", RangePosition::Last) => {
-            source.get_block_number().await.map(|n| n.as_u64()).map_err(|_e| {
+            source.get_block_number().await.map_err(|_e| {
                 ParseError::ParseError("Error retrieving last block number".to_string())
             })
         }
@@ -366,7 +367,7 @@ async fn apply_reorg_buffer(
         0 => Ok(block_chunks),
         reorg_filter => {
             let latest_block = match source.get_block_number().await {
-                Ok(result) => result.as_u64(),
+                Ok(result) => result,
                 Err(_e) => {
                     return Err(ParseError::ParseError("reorg buffer parse error".to_string()))
                 }
@@ -387,24 +388,28 @@ pub(crate) async fn get_latest_block_number(source: Arc<Source>) -> Result<u64,
     source
         .get_block_number()
         .await
-        .map(|n| n.as_u64())
         .map_err(|_e| ParseError::ParseError("Error retrieving latest block number".to_string()))
 }
 
 #[cfg(test)]
 mod tests {
+    use std::path::PathBuf;
+
+    use alloy::{providers::{IpcConnect, ProviderBuilder}, transports::ipc::MockIpcServer};
+
     use super::*;
-    use ethers::prelude::*;
 
+    #[derive(Clone, Debug)]
     enum BlockTokenTest<'a> {
         WithoutMock((&'a str, BlockChunk)),   // Token | Expected
         WithMock((&'a str, BlockChunk, u64)), // Token | Expected | Mock Block Response
     }
 
-    async fn block_token_test_helper(tests: Vec<(BlockTokenTest<'_>, bool)>) {
-        let (provider, mock) = Provider::mocked();
+    async fn block_token_test_helper(tests: Vec<(BlockTokenTest<'_>, bool)>, mock_ipc_path: PathBuf) {
+        let ipc = IpcConnect::new(mock_ipc_path);
+        let provider = ProviderBuilder::new().on_ipc(ipc).await.unwrap().boxed();
         let source = Source {
-            provider: provider.into(),
+            provider,
             semaphore: Arc::new(None),
             rate_limiter: Arc::new(None),
             chain_id: 1,
@@ -416,8 +421,7 @@ mod tests {
         let source = Arc::new(source);
         for (test, res) in tests {
             match test {
-                BlockTokenTest::WithMock((token, expected, latest)) => {
-                    mock.push(U64::from(latest)).unwrap();
+                BlockTokenTest::WithMock((token, expected, _latest)) => {
                     assert_eq!(
                         block_token_test_executor(token, expected, source.clone()).await,
                         res
@@ -458,15 +462,17 @@ mod tests {
         }
     }
 
+    #[derive(Clone, Debug)]
     enum BlockInputTest<'a> {
         WithoutMock((&'a String, Vec<BlockChunk>)), // Token | Expected
         WithMock((&'a String, Vec<BlockChunk>, u64)), // Token | Expected | Mock Block Response
     }
 
-    async fn block_input_test_helper(tests: Vec<(BlockInputTest<'_>, bool)>) {
-        let (provider, mock) = Provider::mocked();
+    async fn block_input_test_helper(tests: Vec<(BlockInputTest<'_>, bool)>, mock_ipc_path: PathBuf) {
+        let ipc = IpcConnect::new(mock_ipc_path);
+        let provider = ProviderBuilder::new().on_ipc(ipc).await.unwrap().boxed();
         let source = Arc::new(Source {
-            provider: provider.into(),
+            provider,
             chain_id: 1,
             rpc_url: "".to_string(),
             inner_request_size: 1,
@@ -477,8 +483,7 @@ mod tests {
         });
         for (test, res) in tests {
             match test {
-                BlockInputTest::WithMock((inputs, expected, latest)) => {
-                    mock.push(U64::from(latest)).unwrap();
+                BlockInputTest::WithMock((inputs, expected, _latest)) => {
                     assert_eq!(
                         block_input_test_executor(inputs, expected, source.clone()).await,
                         res
@@ -531,15 +536,20 @@ mod tests {
         true
     }
 
+    #[derive(Clone, Debug)]
     enum BlockNumberTest<'a> {
         WithoutMock((&'a str, RangePosition, u64)),
         WithMock((&'a str, RangePosition, u64, u64)),
     }
 
-    async fn block_number_test_helper(tests: Vec<(BlockNumberTest<'_>, bool)>) {
-        let (provider, mock) = Provider::mocked();
+    async fn block_number_test_helper(tests: Vec<(BlockNumberTest<'_>, bool)>, mock_ipc_path: PathBuf) {
+        let provider = ProviderBuilder::new()
+            .on_ipc(IpcConnect::new(mock_ipc_path))
+            .await
+            .unwrap()
+            .boxed();
         let source = Source {
-            provider: provider.into(),
+            provider,
             semaphore: Arc::new(None),
             rate_limiter: Arc::new(None),
             chain_id: 1,
@@ -551,8 +561,7 @@ mod tests {
         let source = Arc::new(source);
         for (test, res) in tests {
             match test {
-                BlockNumberTest::WithMock((block_ref, range_position, expected, latest)) => {
-                    mock.push(U64::from(latest)).unwrap();
+                BlockNumberTest::WithMock((block_ref, range_position, expected, _latest)) => {
                     assert_eq!(
                         block_number_test_executor(
                             block_ref,
@@ -604,7 +613,16 @@ mod tests {
             // Number type
             (BlockTokenTest::WithoutMock((r"1", BlockChunk::Numbers(vec![1]))), true), /* Single block */
         ];
-        block_token_test_helper(tests).await;
+        let mut mock_server = MockIpcServer::new();
+        let mock_ipc_path = mock_server.path().clone();
+        for (test,_) in tests.clone().into_iter() {
+            match test {
+                BlockTokenTest::WithoutMock(_) => {}
+                BlockTokenTest::WithMock((_, _, mock_response)) => mock_server.add_reply(mock_response)
+            }
+        }
+        mock_server.spawn().await;
+        block_token_test_helper(tests, mock_ipc_path).await;
     }
 
     #[tokio::test]
@@ -648,7 +666,16 @@ mod tests {
                 true,
             ), // Multi input complex
         ];
-        block_input_test_helper(tests).await;
+        let mut mock_server = MockIpcServer::new();
+        let mock_ipc_path = mock_server.path().clone();
+        for (test, _) in tests.clone() {
+            match test {
+                BlockInputTest::WithMock((_, _, expected)) => mock_server.add_reply(expected),
+                BlockInputTest::WithoutMock(_) => {} 
+            }
+        }
+        mock_server.spawn().await;
+        block_input_test_helper(tests, mock_ipc_path).await;
     }
 
     #[tokio::test]
@@ -666,6 +693,15 @@ mod tests {
             (BlockNumberTest::WithoutMock((r"1m", RangePosition::None, 1000000)), true), // m
             (BlockNumberTest::WithoutMock((r"1k", RangePosition::None, 1000)), true), // k
         ];
-        block_number_test_helper(tests).await;
+        let mut mock_server = MockIpcServer::new();
+        let mock_ipc_path = mock_server.path().clone();
+        for (test, _) in tests.clone() {
+            match test {
+                BlockNumberTest::WithMock((_, _, _, expected)) => mock_server.add_reply(expected),
+                BlockNumberTest::WithoutMock(_) => {} 
+            }
+        }
+        mock_server.spawn().await;
+        block_number_test_helper(tests, mock_ipc_path).await;
     }
 }

crates/cli/src/parse/partitions.rs

@@ -8,7 +8,6 @@ use cryo_freeze::{
     AddressChunk, CallDataChunk, Datatype, Dim, ParseError, Partition, PartitionLabels, SlotChunk,
     Source, Table, TimeDimension, TopicChunk, TransactionChunk,
 };
-use ethers::prelude::*;
 use rand::{seq::SliceRandom, thread_rng};
 use std::{collections::HashMap, str::FromStr, sync::Arc};
 

crates/cli/src/parse/source.rs

@@ -1,36 +1,30 @@
 use std::env;
 
 use crate::args::Args;
-use cryo_freeze::{sources::ProviderWrapper, ParseError, Source, SourceLabels};
-use ethers::prelude::*;
+use alloy::{providers::{IpcConnect, Provider, ProviderBuilder, RootProvider, WsConnect}, transports::{http::reqwest::Url, BoxTransport}};
+use cryo_freeze::{ParseError, Source, SourceLabels};
 use governor::{Quota, RateLimiter};
 use polars::prelude::*;
 use std::num::NonZeroU32;
 
 pub(crate) async fn parse_source(args: &Args) -> Result<Source, ParseError> {
     // parse network info
     let rpc_url = parse_rpc_url(args)?;
-    let (provider, chain_id): (ProviderWrapper, u64) = if rpc_url.starts_with("http") {
-        let provider = Provider::<RetryClient<Http>>::new_client(
-            &rpc_url,
-            args.max_retries,
-            args.initial_backoff,
-        )
-        .map_err(|_e| ParseError::ParseError("could not connect to provider".to_string()))?;
-        let chain_id = provider.get_chainid().await.map_err(ParseError::ProviderError)?.as_u64();
-        (provider.into(), chain_id)
+    let (provider, chain_id): (RootProvider<BoxTransport>, u64) = if rpc_url.starts_with("http") {
+        let url: Url = rpc_url.clone().parse().map_err(ParseError::ParseUrlError)?;
+        let provider = ProviderBuilder::new().on_http(url);
+        let chain_id = provider.get_chain_id().await.map_err(ParseError::ProviderError)?;
+        (provider.boxed(), chain_id)
     } else if rpc_url.starts_with("ws") {
-        let provider = Provider::<Ws>::connect(&rpc_url).await.map_err(|_| {
-            ParseError::ParseError("could not instantiate HTTP Provider".to_string())
-        })?;
-        let chain_id = provider.get_chainid().await.map_err(ParseError::ProviderError)?.as_u64();
-        (provider.into(), chain_id)
+        let ws = WsConnect::new(rpc_url.clone());
+        let provider = ProviderBuilder::new().on_ws(ws).await.map_err(ParseError::ProviderError)?;
+        let chain_id = provider.get_chain_id().await.map_err(ParseError::ProviderError)?;
+        (provider.boxed(), chain_id)
     } else if rpc_url.ends_with(".ipc") {
-        let provider: Provider<Ipc> = Provider::connect_ipc(&rpc_url).await.map_err(|_| {
-            ParseError::ParseError("could not instantiate HTTP Provider".to_string())
-        })?;
-        let chain_id = provider.get_chainid().await.map_err(ParseError::ProviderError)?.as_u64();
-        (provider.into(), chain_id)
+        let ipc = IpcConnect::new(rpc_url.clone());
+        let provider = ProviderBuilder::new().on_ipc(ipc).await.map_err(ParseError::ProviderError)?;
+        let chain_id = provider.get_chain_id().await.map_err(ParseError::ProviderError)?;
+        (provider.boxed(), chain_id)
     } else {
         return Err(ParseError::ParseError(format!("invalid rpc url: {}", rpc_url)));
     };

crates/cli/src/parse/timestamps.rs

@@ -1,3 +1,4 @@
+use alloy::rpc::types::BlockTransactionsKind;
 use cryo_freeze::{BlockChunk, ParseError, Source};
 use polars::prelude::*;
 
@@ -277,31 +278,31 @@ async fn timestamp_to_block_number(timestamp: u64, source: Arc<Source>) -> Resul
     let mut r = latest_block_number;
     let mut mid = (l + r) / 2;
     let mut block = source
-        .get_block(mid)
+        .get_block(mid, BlockTransactionsKind::Hashes)
         .await
         .map_err(|_e| ParseError::ParseError("Error fetching block for timestamp".to_string()))?
         .unwrap();
 
     while l <= r {
         mid = (l + r) / 2;
         block = source
-            .get_block(mid)
+            .get_block(mid, BlockTransactionsKind::Hashes)
             .await
             .map_err(|_e| ParseError::ParseError("Error fetching block for timestamp".to_string()))?
             .unwrap();
 
         #[allow(clippy::comparison_chain)]
-        if block.timestamp == timestamp.into() {
+        if block.header.timestamp == timestamp {
             return Ok(mid);
-        } else if block.timestamp < timestamp.into() {
+        } else if block.header.timestamp < timestamp {
             l = mid + 1;
         } else {
             r = mid - 1;
         }
     }
 
     // If timestamp is between two different blocks, return the lower block.
-    if mid > 0 && block.timestamp > ethers::types::U256::from(timestamp) {
+    if mid > 0 && block.header.timestamp > timestamp {
         Ok(mid - 1)
     } else {
         Ok(mid)
@@ -311,44 +312,40 @@ async fn timestamp_to_block_number(timestamp: u64, source: Arc<Source>) -> Resul
 async fn get_latest_timestamp(source: Arc<Source>) -> Result<u64, ParseError> {
     let latest_block_number = get_latest_block_number(source.clone()).await?;
     let latest_block = source
-        .get_block(latest_block_number)
+        .get_block(latest_block_number, BlockTransactionsKind::Hashes)
         .await
         .map_err(|_e| ParseError::ParseError("Error fetching latest block".to_string()))?
         .unwrap();
 
-    Ok(latest_block.timestamp.as_u64())
+    Ok(latest_block.header.timestamp)
 }
 
 #[cfg(test)]
 mod tests {
     use std::num::NonZeroU32;
 
+    use alloy::providers::ProviderBuilder;
     use governor::{Quota, RateLimiter};
 
     use super::*;
     use cryo_freeze::SourceLabels;
-    use ethers::prelude::*;
 
     async fn setup_source() -> Source {
         let rpc_url = match crate::parse::source::parse_rpc_url(&Args::default()) {
             Ok(url) => url,
             Err(_) => std::process::exit(0),
         };
-        let max_retry = 5;
-        let initial_backoff = 500;
+        // let max_retry = 5;
+        // let initial_backoff = 500;
         let max_concurrent_requests = 100;
-        let provider =
-            Provider::<RetryClient<Http>>::new_client(&rpc_url, max_retry, initial_backoff)
-                .map_err(|_e| ParseError::ParseError("could not connect to provider".to_string()))
-                .unwrap();
-
+        let provider = ProviderBuilder::new().on_http(rpc_url.parse().unwrap()).boxed();
         let quota = Quota::per_second(NonZeroU32::new(15).unwrap())
             .allow_burst(NonZeroU32::new(1).unwrap());
         let rate_limiter = Some(RateLimiter::direct(quota));
         let semaphore = tokio::sync::Semaphore::new(max_concurrent_requests as usize);
 
         Source {
-            provider: provider.into(),
+            provider,
             semaphore: Arc::new(Some(semaphore)),
             rate_limiter: Arc::new(rate_limiter),
             chain_id: 1,
@@ -373,8 +370,8 @@ mod tests {
         let source = setup_source().await;
         let source = Arc::new(source);
         let latest_block_number = get_latest_block_number(source.clone()).await.unwrap();
-        let latest_block = source.get_block(latest_block_number).await.unwrap().unwrap();
-        let latest_timestamp = latest_block.timestamp.as_u64();
+        let latest_block = source.get_block(latest_block_number, BlockTransactionsKind::Hashes).await.unwrap().unwrap();
+        let latest_timestamp = latest_block.header.timestamp;
 
         assert_eq!(
             timestamp_to_block_number(latest_timestamp, source).await.unwrap(),