datafusion_ethers/
stream.rs

1use alloy_network::AnyNetwork;
2use alloy_provider::{DynProvider, Provider};
3use alloy_rpc_types_eth::{BlockNumberOrTag, Filter, FilterBlockOption, Log};
4use alloy_transport::{RpcError, TransportErrorKind};
5
6///////////////////////////////////////////////////////////////////////////////////////////////////
7
8#[derive(Debug, Clone)]
9pub struct StreamOptions {
10    /// Number of blocks to scan in one query. This should be small, as many
11    /// RPC providers impose limits on this parameter
12    pub block_stride: u64,
13
14    /// Many providers don't yet return `blockTimestamp` from `eth_getLogs` RPC endpoint
15    /// and in such cases `block_timestamp` column will be `null`.
16    /// If you enable this fallback the library will perform additional call to `eth_getBlock`
17    /// to populate the timestam, but this may result in significant performance penalty when
18    /// fetching many log records.
19    ///
20    /// See: https://github.com/ethereum/execution-apis/issues/295
21    pub use_block_timestamp_fallback: bool,
22}
23
24impl Default for StreamOptions {
25    fn default() -> Self {
26        Self {
27            block_stride: 10_000,
28            use_block_timestamp_fallback: false,
29        }
30    }
31}
32
33///////////////////////////////////////////////////////////////////////////////////////////////////
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct StreamState {
37    pub last_seen_block: u64,
38}
39
40///////////////////////////////////////////////////////////////////////////////////////////////////
41
42pub struct StreamBatch {
43    pub logs: Vec<Log>,
44    pub state: StreamState,
45    pub block_range_all: (u64, u64),
46}
47
48///////////////////////////////////////////////////////////////////////////////////////////////////
49
50pub struct RawLogsStream;
51
52impl RawLogsStream {
53    // TODO: Re-org detection and handling
54    /// Streams batches of raw logs efficient and resumable pagination over `eth_getLogs` RPC endpoint,
55    pub fn paginate(
56        rpc_client: DynProvider<AnyNetwork>,
57        mut filter: Filter,
58        options: StreamOptions,
59        resume_from_state: Option<StreamState>,
60    ) -> impl futures::Stream<Item = Result<StreamBatch, RpcError<TransportErrorKind>>> {
61        async_stream::try_stream! {
62            // Determine query's full block range, resolving symbolic block aliases like
63            // 'latest' and 'finalized' to block numbers
64            let block_range_all = Self::filter_to_block_range(&rpc_client, &filter.block_option).await?;
65
66            // Subtract from the full block range the range that was already processed
67            let block_range_unprocessed = if let Some(last_seen_block) = resume_from_state.map(|s| s.last_seen_block) {
68                (
69                    u64::max(last_seen_block + 1, block_range_all.0),
70                    block_range_all.1,
71                )
72            } else {
73                block_range_all
74            };
75
76            tracing::info!(
77                block_range_query = ?filter.block_option,
78                ?block_range_all,
79                ?block_range_unprocessed,
80                "Computed block ranges",
81            );
82
83            let mut block_range_to_scan = block_range_unprocessed;
84
85            while block_range_to_scan.0 <= block_range_to_scan.1 {
86                let block_range_page = (
87                    block_range_to_scan.0,
88                    u64::min(
89                        block_range_to_scan.1,
90                        block_range_to_scan.0 + options.block_stride - 1,
91                    ),
92                );
93
94                // Setup per-query filter
95                filter.block_option = FilterBlockOption::Range {
96                    from_block: Some(block_range_page.0.into()),
97                    to_block: Some(block_range_page.1.into()),
98                };
99
100                tracing::debug!(
101                    ?block_range_page,
102                    "Querying block range",
103                );
104
105                // Query the node
106                let mut logs = rpc_client.get_logs(&filter).await?;
107
108                // Resolve timestamp if needed
109                if options.use_block_timestamp_fallback && logs.iter().any(|l| l.block_timestamp.is_none()) {
110                    Self::populate_block_timestamps_fallback(&rpc_client, &mut logs).await?;
111                }
112
113                yield StreamBatch {
114                    logs,
115                    state: StreamState { last_seen_block: block_range_page.1 },
116                    block_range_all,
117                };
118
119                // Update remaining range
120                block_range_to_scan.0 = block_range_page.1 + 1;
121            }
122        }
123    }
124
125    /// This function will determine the timestamps for the first and last block in the batch
126    /// and approximates the timestamps for blocks in between by interpolating.
127    /// NOTE: This might not be reproducible if stride changes between runs
128    async fn populate_block_timestamps_fallback(
129        rpc_client: &DynProvider<AnyNetwork>,
130        logs: &mut [Log],
131    ) -> Result<(), RpcError<TransportErrorKind>> {
132        let first_log = logs.first().unwrap();
133        let last_log = logs.last().unwrap();
134
135        let first_block = rpc_client
136            .get_block_by_hash(first_log.block_hash.unwrap())
137            .await?
138            .unwrap();
139
140        let last_block = if first_log.block_hash == last_log.block_hash {
141            first_block.clone()
142        } else {
143            rpc_client
144                .get_block_by_hash(last_log.block_hash.unwrap())
145                .await?
146                .unwrap()
147        };
148
149        let time_per_block = ((last_block.header.timestamp - first_block.header.timestamp) as f64)
150            / ((last_block.number() - first_block.number()) as f64);
151
152        tracing::debug!(
153            block_range = ?(first_block.number(), last_block.number()),
154            time_per_block,
155            "Populating block timestamps via fallback mechanism",
156        );
157
158        for log in logs.iter_mut() {
159            let block_offset = log.block_number.unwrap() - first_block.number();
160            let time_offset = time_per_block * (block_offset as f64);
161            let ts = first_block.header.timestamp + (time_offset.floor() as u64);
162            log.block_timestamp = Some(ts);
163        }
164
165        Ok(())
166    }
167
168    pub async fn filter_to_block_range(
169        rpc_client: &DynProvider<AnyNetwork>,
170        block_option: &FilterBlockOption,
171    ) -> Result<(u64, u64), RpcError<TransportErrorKind>> {
172        match block_option {
173            FilterBlockOption::Range {
174                from_block: Some(from),
175                to_block: Some(to),
176            } => {
177                let from = match from {
178                    BlockNumberOrTag::Earliest => 0,
179                    BlockNumberOrTag::Number(n) => *n,
180                    _ => Err(RpcError::local_usage_str(&format!(
181                        "Invalid range: {block_option:?}"
182                    )))?,
183                };
184                let to = match to {
185                    BlockNumberOrTag::Number(n) => *n,
186                    BlockNumberOrTag::Latest
187                    | BlockNumberOrTag::Safe
188                    | BlockNumberOrTag::Finalized => {
189                        let Some(to_block) = rpc_client.get_block((*to).into()).await? else {
190                            Err(RpcError::local_usage_str(&format!(
191                                "Unable to resolve block: {to:?}"
192                            )))?
193                        };
194                        to_block.header.number
195                    }
196                    _ => Err(RpcError::local_usage_str(&format!(
197                        "Invalid range: {block_option:?}"
198                    )))?,
199                };
200                Ok((from, to))
201            }
202            FilterBlockOption::Range { .. } => Err(RpcError::local_usage_str(&format!(
203                "Invalid range: {block_option:?}"
204            )))?,
205            FilterBlockOption::AtBlockHash(_) => {
206                unimplemented!("Querying a single block by hash is not yet supported")
207            }
208        }
209    }
210}
211
212///////////////////////////////////////////////////////////////////////////////////////////////////
datafusion_ethers/stream.rs

datafusion_ethers/
stream.rs
