ethl/lib.rs
1//! The ethl crate provides tools for capturing, processing, archiving, and replaying Ethereum events. The impetus for this
2//! crate was to create an opinionated way to archive Ethereum logs to Parquet files in a way that is efficient for both
3//! storage, querying, and replaying deserialized data for downstream processing.
4//!
5//! # Storage Model
6//!
7//! Event data is stored in Parquet files, where the schema is derived from the Solidity event definition. Each events data
8//! is stored in its own Parquet file group, which allows for efficient querying and retrieval of specific event types, and makes it easy to
9//! add new event types without impacting existing data.
10//!
11//! Core assumptions:
12//! - Events are stored in immutable Parquet files where the filename includes the block range covered by the file
13//! - Events are written in block number order, and within a block, in log index order
14//! - Events are stored in a directory structure that is compatible with S3 and local filesystem backends
15//! - Event parquet files are stored in a directory named after the event name along with the first 4 bytes of the event signature hash
16//! - Within a directory, parquet files are named based on the block range they cover (eg: 000000000001-000000000200.parquet)
17//! - The schema for each event is derived from the Solidity event definition, and includes all indexed and non-indexed parameters
18//! - The schema also includes metadata columns for block number, log index, and address
19//! - Schemas favor human friendly types (eg: strings for addresses and uint256) over compact types (size delta wes negligible, perf unknown)
20//!
21//! Example directory structure:
22//! ```text
23//! /path/to/archive/
24//! ├── transfer_0xdeadbeef/
25//! │ ├── 000000000001-000000000200.parquet
26//! │ ├── 000000000201-000000000400.parquet
27//! │ └── ...
28//! ├── approval_0xbeeffeee/
29//! │ ├── 000000000001-000000000200.parquet
30//! │ ├── 000000000201-000000000400.parquet
31//! │ └── ...
32//! └── ...
33//! ```
34//!
35//! Example schema for a swap event:
36//! ```text
37//! message arrow_schema {
38//! REQUIRED INT64 log_block (INTEGER(64,false));
39//! REQUIRED INT32 log_index (INTEGER(32,false));
40//! REQUIRED BYTE_ARRAY log_address (STRING);
41//! REQUIRED BYTE_ARRAY sender (STRING);
42//! REQUIRED BYTE_ARRAY recipient (STRING);
43//! REQUIRED BYTE_ARRAY amount0 (STRING);
44//! REQUIRED BYTE_ARRAY amount1 (STRING);
45//! REQUIRED BYTE_ARRAY sqrtPriceX96 (STRING);
46//! REQUIRED BYTE_ARRAY liquidity (STRING);
47//! REQUIRED INT32 tick;
48//! REQUIRED BYTE_ARRAY protocolFeesToken0 (STRING);
49//! REQUIRED BYTE_ARRAY protocolFeesToken1 (STRING);
50//! }
51//! ```
52//!
53//! # Archiving Events and Loading Events
54//!
55//! The main entry point for archiving and loading events is the `EventArchive` struct, which provides methods to advance the archive to a specific block or to the latest block.
56//! It also provides a method to create an `EventLoader` for reading archived events.
57//!
58//! Example usage:
59//! ```no_run
60//! use ethl::{archive::EventArchive, rpc::config::ProviderSettings};
61//! use alloy::json_abi::Event;
62//! use anyhow::Result;
63//! use tracing::info;
64//! use futures_util::{StreamExt, pin_mut};
65//!
66//! #[tokio::main]
67//! async fn main() -> Result<()> {
68//! let providers = ProviderSettings::default();
69//! let events = vec![
70//! Event::parse("Transfer(address indexed from, address indexed to, uint256 value)")?,
71//! Event::parse("Approval(address indexed owner, address indexed spender, uint256 value)")?
72//! ];
73//! let archive = EventArchive::from_uri("s3://my-bucket/events", &providers, events)?;
74//! let latest_block = archive.advance_to_latest().await?;
75//! info!("Archived events to block {}", latest_block);
76//!
77//! let loader = archive.loader()?;
78//! let stream = loader.load_range(None, None).await;
79//! pin_mut!(stream);
80//! while let Some(block_events) = stream.next().await {
81//! let block_events = block_events?;
82//! info!("Block: {}", block_events.block_number);
83//! for event in block_events.events {
84//! info!(" Event: {:?}", event);
85//! }
86//! }
87//! Ok(())
88//! }
89//! ```
90//!
91//!
92//! # Extracting Events from RPC Nodes
93//!
94//! Events are extracted from RPC nodes and written to Parquet files via `EventExtractor`, which automatically generates the schemas for decoded events
95//!
96//! Example usage:
97//! ```no_run
98//! use ethl::{archive::extract::EventExtractor, rpc::config::ProviderSettings};
99//! use alloy::json_abi::Event;
100//! use alloy::rpc::types::Filter;
101//! use anyhow::Result;
102//!
103//! #[tokio::main]
104//! async fn main() -> Result<()> {
105//! let providers = ProviderSettings::default();
106//!
107//! // This can also be captured from the SolEvent::abi() method if using alloy-json-abi
108//! let events = vec![
109//! Event::parse("Transfer(address indexed from, address indexed to, uint256 value)")?,
110//! Event::parse("Approval(address indexed owner, address indexed spender, uint256 value)")?
111//! ];
112//!
113//! // EventExtractor will automatically add events and to_block (if not set) to the filter
114//! let filter = Filter::new()
115//! .from_block(10000000u64);
116//! let mut extractor = EventExtractor::from_uri(
117//! "s3://my-bucket/events",
118//! 200_000,
119//! providers,
120//! events,
121//! filter,
122//! )?;
123//! extractor.extract().await?;
124//! Ok(())
125//! }
126//! ```
127//!
128//! # Reading Events
129//! Archived events are read via `EventLoader`, which merges events by block, sorted by log index from stored Parquet files.
130//!
131//! Example usage:
132//! ```no_run
133//! use ethl::archive::load::EventLoader;
134//! use alloy::json_abi::Event;
135//! use anyhow::Result;
136//! use futures_util::{StreamExt, pin_mut};
137//! use tracing::info;
138//!
139//! #[tokio::main]
140//! async fn main() -> Result<()> {
141//! let events = vec![
142//! Event::parse("Transfer(address indexed from, address indexed to, uint256 value)")?,
143//! Event::parse("Approval(address indexed owner, address indexed spender, uint256 value)")?
144//! ];
145//! let loader = EventLoader::from_uri("s3://my-bucket/events", &events)?;
146//! let stream = loader.load_range(None, None).await;
147//! pin_mut!(stream);
148//! while let Some(block_events) = stream.next().await {
149//! let block_events = block_events?;
150//! info!("Block: {}", block_events.block_number);
151//! for event in block_events.events {
152//! info!(" Event: {:?}", event);
153//! }
154//! }
155//! Ok(())
156//! }
157//! ```
158//!
159
160pub mod archive;
161pub mod rpc;
162pub mod storage;