chore: migrate from alloy 0.6.4 to 1.0.9

This migration updates the codebase to use alloy 1.0.9, which includes several breaking API changes:

- Provider architecture changes: BoxTransport replaced with DynProvider
- Transaction field access: tx.from replaced with tx.inner.signer()
- Trace API methods now use builder pattern with .trace_types()
- Updated deprecated methods (on_http -> connect_http, etc.)
- Removed boolean parameters from log decoding methods
- Fixed type parameters and imports throughout the codebase

All tests pass after migration.

Author: mattsse

Cargo.lock

@@ -19,7 +19,7 @@ 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.6.4", features = [
+alloy = { version = "1.0.9", features = [
     "full",
     "rpc-types-trace",
     "provider-ws",

Cargo.toml

@@ -395,8 +395,8 @@ mod tests {
     use std::path::PathBuf;
 
     use alloy::{
-        providers::{IpcConnect, ProviderBuilder},
-        transports::ipc::MockIpcServer,
+        providers::{Provider, ProviderBuilder},
+        transports::ipc::{IpcConnect, MockIpcServer},
     };
 
     use super::*;
@@ -412,7 +412,7 @@ mod tests {
         mock_ipc_path: PathBuf,
     ) {
         let ipc = IpcConnect::new(mock_ipc_path);
-        let provider = ProviderBuilder::new().on_ipc(ipc).await.unwrap().boxed();
+        let provider = ProviderBuilder::new().connect_ipc(ipc).await.unwrap().erased();
         let source = Source {
             provider,
             semaphore: Arc::new(None),
@@ -478,7 +478,7 @@ mod tests {
         mock_ipc_path: PathBuf,
     ) {
         let ipc = IpcConnect::new(mock_ipc_path);
-        let provider = ProviderBuilder::new().on_ipc(ipc).await.unwrap().boxed();
+        let provider = ProviderBuilder::new().connect_ipc(ipc).await.unwrap().erased();
         let source = Arc::new(Source {
             provider,
             chain_id: 1,
@@ -554,8 +554,11 @@ mod tests {
         tests: Vec<(BlockNumberTest<'_>, bool)>,
         mock_ipc_path: PathBuf,
     ) {
-        let provider =
-            ProviderBuilder::new().on_ipc(IpcConnect::new(mock_ipc_path)).await.unwrap().boxed();
+        let provider = ProviderBuilder::new()
+            .connect_ipc(IpcConnect::new(mock_ipc_path))
+            .await
+            .unwrap()
+            .erased();
         let source = Source {
             provider,
             semaphore: Arc::new(None),

crates/cli/src/parse/blocks.rs

@@ -1,14 +1,13 @@
-use std::env;
+use std::{env, sync::Arc};
 
 use crate::args::Args;
 use alloy::{
-    providers::{Provider, ProviderBuilder, RootProvider},
-    rpc::client::{BuiltInConnectionString, ClientBuilder, RpcClient},
-    transports::{layers::RetryBackoffLayer, BoxTransport},
+    providers::{Provider, ProviderBuilder},
+    rpc::client::ClientBuilder,
+    transports::layers::RetryBackoffLayer,
 };
 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> {
@@ -19,14 +18,12 @@ pub(crate) async fn parse_source(args: &Args) -> Result<Source, ParseError> {
         args.initial_backoff,
         args.compute_units_per_second,
     );
-    let connect: BuiltInConnectionString = rpc_url.parse().map_err(ParseError::ProviderError)?;
-    let client: RpcClient<BoxTransport> = ClientBuilder::default()
+    let client = ClientBuilder::default()
         .layer(retry_layer)
-        .connect_boxed(connect)
+        .connect(&rpc_url)
         .await
-        .map_err(ParseError::ProviderError)?
-        .boxed();
-    let provider: RootProvider<BoxTransport> = ProviderBuilder::default().on_client(client);
+        .map_err(ParseError::ProviderError)?;
+    let provider = ProviderBuilder::default().connect_client(client).erased();
     let chain_id = provider.get_chain_id().await.map_err(ParseError::ProviderError)?;
     let rate_limiter = match args.requests_per_second {
         Some(rate_limit) => match (NonZeroU32::new(1), NonZeroU32::new(rate_limit)) {

crates/cli/src/parse/source.rs

@@ -325,9 +325,9 @@ mod tests {
     use std::num::NonZeroU32;
 
     use alloy::{
-        providers::ProviderBuilder,
-        rpc::client::{BuiltInConnectionString, ClientBuilder, RpcClient},
-        transports::{layers::RetryBackoffLayer, BoxTransport},
+        providers::{Provider, ProviderBuilder},
+        rpc::client::ClientBuilder,
+        transports::layers::RetryBackoffLayer,
     };
     use governor::{Quota, RateLimiter};
 
@@ -345,16 +345,13 @@ mod tests {
         let max_concurrent_requests = 100;
         let retry_layer =
             RetryBackoffLayer::new(max_retry, initial_backoff, compute_units_per_second);
-        let connect: BuiltInConnectionString =
-            rpc_url.parse().map_err(ParseError::ProviderError).unwrap();
-        let client: RpcClient<BoxTransport> = ClientBuilder::default()
+        let client = ClientBuilder::default()
             .layer(retry_layer)
-            .connect_boxed(connect)
+            .connect(&rpc_url)
             .await
             .map_err(ParseError::ProviderError)
-            .unwrap()
-            .boxed();
-        let provider = ProviderBuilder::default().on_client(client);
+            .unwrap();
+        let provider = ProviderBuilder::default().connect_client(client).erased();
         let quota = Quota::per_second(NonZeroU32::new(15).unwrap())
             .allow_burst(NonZeroU32::new(1).unwrap());
         let rate_limiter = Some(RateLimiter::direct(quota));

crates/cli/src/parse/timestamps.rs

@@ -136,7 +136,7 @@ fn process_logs(logs: Vec<Log>, columns: &mut Logs, schema: &Table) -> R<()> {
             if let (Some(decoder), Some(indexed_keys), Some(body_keys)) =
                 (&schema.log_decoder, &indexed_keys, &body_keys)
             {
-                match decoder.event.decode_log(&log.inner.data, true) {
+                match decoder.event.decode_log(&log.inner.data) {
                     Ok(log) => {
                         // for param in log.indexed {
                         //     if decode_keys.contains(param.name.as_str()) {

crates/freeze/src/datasets/logs.rs

@@ -93,7 +93,7 @@ impl CollectByBlock for Transactions {
         // filter by from_address
         let from_filter: Box<dyn Fn(&Transaction) -> bool + Send> =
             if let Some(from_address) = &request.from_address {
-                Box::new(move |tx| tx.from == Address::from_slice(from_address))
+                Box::new(move |tx| tx.inner.signer() == Address::from_slice(from_address))
             } else {
                 Box::new(|_| true)
             };
@@ -228,7 +228,7 @@ pub(crate) fn process_transaction(
     store!(schema, columns, block_number, tx.block_number.map(|x| x as u32));
     store!(schema, columns, transaction_index, tx.transaction_index);
     store!(schema, columns, transaction_hash, tx.inner.tx_hash().to_vec());
-    store!(schema, columns, from_address, tx.from.to_vec());
+    store!(schema, columns, from_address, tx.inner.signer().to_vec());
     store!(
         schema,
         columns,
@@ -277,15 +277,15 @@ pub(crate) fn process_transaction(
 }
 
 fn get_max_fee_per_gas(tx: &Transaction) -> Option<u64> {
-    match &tx.inner {
+    match tx.inner.as_ref() {
         alloy::consensus::TxEnvelope::Legacy(_) => None,
         alloy::consensus::TxEnvelope::Eip2930(_) => None,
         _ => Some(tx.inner.max_fee_per_gas() as u64),
     }
 }
 
 pub(crate) fn get_gas_price(block: &Block, tx: &Transaction) -> Option<u64> {
-    match &tx.inner {
+    match tx.inner.as_ref() {
         alloy::consensus::TxEnvelope::Legacy(_) => tx.gas_price().map(|gas_price| gas_price as u64),
         alloy::consensus::TxEnvelope::Eip2930(_) => {
             tx.gas_price().map(|gas_price| gas_price as u64)

crates/freeze/src/datasets/transactions.rs

@@ -62,8 +62,7 @@ impl LogDecoder {
             .collect();
 
         for log in logs {
-            match self.event.decode_log_parts(log.topics().to_vec(), log.data().data.as_ref(), true)
-            {
+            match self.event.decode_log_parts(log.topics().to_vec(), log.data().data.as_ref()) {
                 Ok(decoded) => {
                     for (idx, param) in decoded.indexed.into_iter().enumerate() {
                         map.entry(indexed_keys[idx].clone()).or_default().push(param);

crates/freeze/src/types/decoders/log_decoder.rs

@@ -59,7 +59,7 @@ pub use datatypes::*;
 pub use files::{ColumnEncoding, FileFormat, FileOutput, SubDir};
 pub use queries::{Query, QueryLabels, TimeDimension};
 pub use schemas::{ColumnType, SchemaFunctions, Schemas, Table, U256Type};
-pub use sources::{Fetcher, RateLimiter, Source, SourceLabels};
+pub use sources::{RateLimiter, Source, SourceLabels};
 // pub(crate) use summaries::FreezeSummaryAgg;
 // pub use summaries::{FreezeChunkSummary, FreezeSummary};
 pub use summaries::{print_all_datasets, print_dataset_info, FreezeSummary};

crates/freeze/src/types/mod.rs

@@ -5,7 +5,7 @@ use alloy::{
     primitives::{Address, BlockNumber, Bytes, TxHash, B256, U256},
     providers::{
         ext::{DebugApi, TraceApi},
-        Provider, ProviderBuilder, RootProvider,
+        DynProvider, Provider, ProviderBuilder,
     },
     rpc::types::{
         trace::{
@@ -22,7 +22,7 @@ use alloy::{
         Block, BlockTransactions, BlockTransactionsKind, Filter, Log, Transaction,
         TransactionInput, TransactionReceipt, TransactionRequest,
     },
-    transports::{http::reqwest::Url, BoxTransport, RpcError, TransportErrorKind},
+    transports::{http::reqwest::Url, RpcError, TransportErrorKind},
 };
 use governor::{
     clock::DefaultClock,
@@ -43,7 +43,7 @@ pub type RateLimiter = governor::RateLimiter<NotKeyed, InMemoryState, DefaultClo
 #[derive(Clone, Debug)]
 pub struct Source {
     /// provider
-    pub provider: RootProvider<BoxTransport>,
+    pub provider: DynProvider,
     /// chain_id of network
     pub chain_id: u64,
     /// number of blocks per log request
@@ -119,7 +119,7 @@ impl Source {
     pub async fn init(rpc_url: Option<String>) -> Result<Source> {
         let rpc_url: String = parse_rpc_url(rpc_url);
         let parsed_rpc_url: Url = rpc_url.parse().expect("rpc url is not valid");
-        let provider = ProviderBuilder::new().on_http(parsed_rpc_url.clone());
+        let provider = ProviderBuilder::new().connect_http(parsed_rpc_url.clone());
         let chain_id = provider
             .get_chain_id()
             .await
@@ -128,10 +128,10 @@ impl Source {
         let rate_limiter = None;
         let semaphore = None;
 
-        let provider = ProviderBuilder::new().on_http(parsed_rpc_url);
+        let provider = ProviderBuilder::new().connect_http(parsed_rpc_url);
 
         let source = Source {
-            provider: provider.boxed(),
+            provider: provider.erased(),
             chain_id,
             inner_request_size: DEFAULT_INNER_REQUEST_SIZE,
             max_concurrent_chunks: Some(DEFAULT_MAX_CONCURRENT_CHUNKS),
@@ -210,17 +210,6 @@ pub struct SourceLabels {
     pub initial_backoff: Option<u64>,
 }
 
-/// Wrapper over `Provider<P>` that adds concurrency and rate limiting controls
-#[derive(Debug)]
-pub struct Fetcher<P> {
-    /// provider data source
-    pub provider: RootProvider<P>,
-    /// semaphore for controlling concurrency
-    pub semaphore: Option<Semaphore>,
-    /// rate limiter for controlling request rate
-    pub rate_limiter: Option<RateLimiter>,
-}
-
 type Result<T> = ::core::result::Result<T, CollectError>;
 
 // impl<P: JsonRpcClient> Fetcher<P> {
@@ -238,9 +227,12 @@ impl Source {
         trace_types: Vec<TraceType>,
     ) -> Result<Vec<TraceResultsWithTransactionHash>> {
         let _permit = self.permit_request().await;
-        Self::map_err(
-            self.provider.trace_replay_block_transactions(block.into(), &trace_types).await,
-        )
+        let trace_result = self
+            .provider
+            .trace_replay_block_transactions(block.into())
+            .trace_types(trace_types)
+            .await;
+        Self::map_err(trace_result)
     }
 
     /// Get state diff traces of block
@@ -298,7 +290,9 @@ impl Source {
         trace_types: Vec<TraceType>,
     ) -> Result<TraceResults> {
         let _permit = self.permit_request().await;
-        Self::map_err(self.provider.trace_replay_transaction(tx_hash, &trace_types).await)
+        let trace_result =
+            self.provider.trace_replay_transaction(tx_hash).trace_types(trace_types).await;
+        Self::map_err(trace_result)
     }
 
     /// Get state diff traces of transaction
@@ -348,7 +342,11 @@ impl Source {
         kind: BlockTransactionsKind,
     ) -> Result<Option<Block>> {
         let _permit = self.permit_request().await;
-        Self::map_err(self.provider.get_block(block_num.into(), kind).await)
+        let block_result = match kind {
+            BlockTransactionsKind::Full => self.provider.get_block(block_num.into()).await,
+            BlockTransactionsKind::Hashes => self.provider.get_block(block_num.into()).await,
+        };
+        Self::map_err(block_result)
     }
 
     /// Gets the block with `block_hash` (transaction hashes only)
@@ -358,7 +356,11 @@ impl Source {
         kind: BlockTransactionsKind,
     ) -> Result<Option<Block>> {
         let _permit = self.permit_request().await;
-        Self::map_err(self.provider.get_block(block_hash.into(), kind).await)
+        let block_result = match kind {
+            BlockTransactionsKind::Full => self.provider.get_block(block_hash.into()).await,
+            BlockTransactionsKind::Hashes => self.provider.get_block(block_hash.into()).await,
+        };
+        Self::map_err(block_result)
     }
 
     /// Returns all receipts for a block.
@@ -398,7 +400,7 @@ impl Source {
         block_number: BlockNumber,
     ) -> Result<Bytes> {
         let _permit = self.permit_request().await;
-        Self::map_err(self.provider.call(&transaction).block(block_number.into()).await)
+        Self::map_err(self.provider.call(transaction).block(block_number.into()).await)
     }
 
     /// Returns traces for given call data
@@ -411,10 +413,14 @@ impl Source {
         let _permit = self.permit_request().await;
         if let Some(bn) = block_number {
             return Self::map_err(
-                self.provider.trace_call(&transaction, &trace_type).block_id(bn.into()).await,
+                self.provider
+                    .trace_call(&transaction)
+                    .trace_types(trace_type.clone())
+                    .block_id(bn.into())
+                    .await,
             );
         }
-        Self::map_err(self.provider.trace_call(&transaction, &trace_type).await)
+        Self::map_err(self.provider.trace_call(&transaction).trace_types(trace_type.clone()).await)
     }
 
     /// Get nonce of address
@@ -495,7 +501,7 @@ impl Source {
             ..Default::default()
         };
         let _permit = self.permit_request().await;
-        Self::map_err(self.provider.call(&transaction).block(block_number.into()).await)
+        Self::map_err(self.provider.call(transaction).block(block_number.into()).await)
     }
 
     /// Return output data of a contract call
@@ -515,12 +521,15 @@ impl Source {
         if block_number.is_some() {
             Self::map_err(
                 self.provider
-                    .trace_call(&transaction, &trace_type)
+                    .trace_call(&transaction)
+                    .trace_types(trace_type.clone())
                     .block_id(block_number.unwrap().into())
                     .await,
             )
         } else {
-            Self::map_err(self.provider.trace_call(&transaction, &trace_type).await)
+            Self::map_err(
+                self.provider.trace_call(&transaction).trace_types(trace_type.clone()).await,
+            )
         }
     }