ethl/storage/codec/
mod.rs

1use std::sync::Arc;
2
3use crate::storage::codec::{decoder::BatchEventDecoder, encoder::BatchEventEncoder};
4use alloy::{dyn_abi::DynSolValue, hex::ToHexExt, json_abi::Event};
5use anyhow::Result;
6use arrow::{
7    array::{Array, ArrayRef},
8    datatypes::SchemaRef,
9};
10use parquet::{arrow::ArrowSchemaConverter, schema::types::SchemaDescriptor};
11
12pub mod decoder;
13pub mod encoder;
14pub mod schema;
15
16// Accumulator for Solidity types to Arrow data arrays (used for accumulating event data in a RecordBatch)
17pub trait SolArrayBuilder {
18    fn append_value(&mut self, value: &DynSolValue);
19    fn len(&self) -> usize;
20    fn finish(&mut self) -> Arc<dyn Array>;
21    fn is_empty(&self) -> bool {
22        self.len() == 0
23    }
24}
25
26// Dynamic reader for Solidity arrays
27pub trait SolArrayReader {
28    fn get(&self, col: &ArrayRef, index: usize) -> DynSolValue;
29}
30
31pub struct SolEventCodec {
32    pub event: Event,
33    pub schema: SchemaRef,
34    pub encoder: BatchEventEncoder,
35    pub decoder: BatchEventDecoder,
36}
37
38impl SolEventCodec {
39    pub fn new(event: &Event) -> Result<Self> {
40        let schema = schema::create_schema(event);
41        let decoder = BatchEventDecoder::new(event);
42        let encoder = BatchEventEncoder::new(event, schema.clone())?;
43        Ok(SolEventCodec {
44            event: event.clone(),
45            schema,
46            encoder,
47            decoder,
48        })
49    }
50
51    pub fn new_encoder(&self) -> BatchEventEncoder {
52        BatchEventEncoder::new(&self.event, self.schema.clone()).unwrap()
53    }
54
55    pub fn event_id(&self) -> String {
56        format!(
57            "{}-{}",
58            self.event.name.to_lowercase(),
59            &self.event.selector().encode_hex_with_prefix()[0..10]
60        )
61    }
62
63    // pub fn path(&self, base_uri: impl AsRef<str>) -> String {
64    //     format!("{}/{}", base_uri.as_ref(), self.event_id())
65    // }
66
67    pub fn parquet_schema(&self) -> SchemaDescriptor {
68        ArrowSchemaConverter::new()
69            .with_coerce_types(false)
70            .convert(&self.schema)
71            .expect("Failed to convert Arrow schema to Parquet schema")
72    }
73}
74
75impl TryInto<SolEventCodec> for &Event {
76    type Error = anyhow::Error;
77    fn try_into(self) -> Result<SolEventCodec> {
78        SolEventCodec::new(self)
79    }
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85    use alloy::{sol, sol_types::JsonAbiExt};
86
87    sol! {
88        #[sol(abi)]
89        #[derive(Debug, Default)]
90        event PoolCreated(address indexed token0, address indexed token1, bool indexed stable, address pool, uint256 count);
91        #[sol(abi)]
92        event SetCustomFee(address indexed pool, uint256 fee);
93    }
94
95    #[test]
96    fn test_sol_event_codec() -> Result<()> {
97        let codec = SolEventCodec::new(&PoolCreated::abi())?;
98        assert_eq!(codec.schema.fields().len(), 8); // 3 header + 5 event params
99        Ok(())
100    }
101
102    /// Regression test for BigIntConversionError panic when encoding Mint events
103    /// from non-standard pools (e.g. unsupported factory on Base).
104    ///
105    /// Tx: 0x3acf6cb136fa7f1756f5e699c8d8056ecc0748f4b92ee00431f0e14a6ce03f09
106    /// Pool: 0x0ed46f2668a6394ffa806713a306a51b998eb070 (unsupported factory)
107    ///
108    /// The int24 tickLower/tickUpper fields decoded to values that overflowed
109    /// when using the panicking `as_i*()` conversion on I256.
110    #[test]
111    fn test_encoder_no_panic_on_oversized_int_values() -> Result<()> {
112        use alloy::primitives::{Address, B256, LogData};
113        use alloy::rpc::types::Log;
114
115        sol! {
116            #[sol(abi)]
117            event Mint(address sender, address indexed owner, int24 indexed tickLower, int24 indexed tickUpper, uint128 amount, uint256 amount0, uint256 amount1);
118        }
119
120        let event = Mint::abi();
121        let schema = super::schema::create_schema(&event);
122        let mut encoder = super::encoder::BatchEventEncoder::new(&event, schema)?;
123
124        // Construct a Mint log reproducing the real crash. The bad pool stuffed a
125        // 160-bit address into the indexed int24 tickLower topic, which overflows
126        // i32 and previously caused a BigIntConversionError panic in as_i32().
127        //
128        // Non-indexed fields: sender(address), amount(uint128), amount0(uint256), amount1(uint256)
129        let data_bytes = [0u8; 128]; // 4 x 32-byte zero words
130
131        let log = Log {
132            inner: alloy::primitives::Log {
133                address: "0x0ed46f2668a6394ffa806713a306a51b998eb070"
134                    .parse::<Address>()
135                    .unwrap(),
136                data: LogData::new(
137                    vec![
138                        // topic[0]: Mint event selector
139                        "0x7a53080ba414158be7ec69b987b5fb7d07dee101fe85488f0853ae16239d0bde"
140                            .parse::<B256>()
141                            .unwrap(),
142                        // topic[1]: owner (indexed address)
143                        "0x000000000000000000000000a9b1758a3e74e440c774e4cd7d7c00d9c5d27859"
144                            .parse::<B256>()
145                            .unwrap(),
146                        // topic[2]: tickLower (indexed int24) - address stuffed in, overflows i32
147                        "0x000000000000000000000000a9b1758a3e74e440c774e4cd7d7c00d9c5d27859"
148                            .parse::<B256>()
149                            .unwrap(),
150                        // topic[3]: tickUpper (indexed int24) - valid small value
151                        "0x0000000000000000000000000000000000000000000000000000000000032640"
152                            .parse::<B256>()
153                            .unwrap(),
154                    ],
155                    data_bytes.into(),
156                )
157                .unwrap(),
158            },
159            block_hash: None,
160            block_number: Some(42601282),
161            block_timestamp: None,
162            transaction_hash: None,
163            transaction_index: None,
164            log_index: Some(429),
165            removed: false,
166        };
167
168        // This previously panicked with BigIntConversionError on as_i32()
169        encoder.append(&log);
170        assert_eq!(encoder.len(), 1);
171
172        let batch = encoder.finish()?;
173        assert_eq!(batch.num_rows(), 1);
174        Ok(())
175    }
176}
ethl/storage/codec/mod.rs

ethl/storage/codec/
mod.rs
