anvil/
filter.rs

1//! Support for polling based filters
2use crate::{
3    StorageInfo,
4    eth::{
5        backend::notifications::{ChainNotification, ChainNotifications},
6        error::ToRpcResponseResult,
7    },
8    pubsub::{filter_logs, filter_removed_logs},
9};
10use alloy_consensus::TxReceipt;
11use alloy_network::{AnyRpcTransaction, Network};
12use alloy_primitives::{TxHash, map::HashMap};
13use alloy_rpc_types::{Filter, FilteredParams, Log};
14use anvil_core::eth::subscription::SubscriptionId;
15use anvil_rpc::{
16    error::{ErrorCode, RpcError},
17    response::ResponseResult,
18};
19use futures::{Stream, StreamExt, channel::mpsc::Receiver};
20use std::{
21    pin::Pin,
22    sync::Arc,
23    task::{Context, Poll},
24    time::{Duration, Instant},
25};
26use tokio::sync::{Mutex, mpsc};
27
28/// Type alias for filters identified by their id and their expiration timestamp
29type FilterMap<N> = Arc<Mutex<HashMap<String, (EthFilter<N>, Instant)>>>;
30
31/// timeout after which to remove an active filter if it wasn't polled since then
32pub const ACTIVE_FILTER_TIMEOUT_SECS: u64 = 60 * 5;
33
34/// Contains all registered filters
35pub struct Filters<N: Network> {
36    /// all currently active filters
37    active_filters: FilterMap<N>,
38    /// How long we keep a live the filter after the last poll
39    keepalive: Duration,
40}
41
42impl<N: Network> Clone for Filters<N> {
43    fn clone(&self) -> Self {
44        Self { active_filters: self.active_filters.clone(), keepalive: self.keepalive }
45    }
46}
47
48impl<N: Network> std::fmt::Debug for Filters<N> {
49    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
50        f.debug_struct("Filters").field("keepalive", &self.keepalive).finish_non_exhaustive()
51    }
52}
53
54impl<N: Network> Filters<N> {
55    /// Adds a new `EthFilter` to the set
56    pub async fn add_filter(&self, filter: EthFilter<N>) -> String {
57        let id = new_id();
58        trace!(target: "node::filter", "Adding new filter id {}", id);
59        let mut filters = self.active_filters.lock().await;
60        filters.insert(id.clone(), (filter, self.next_deadline()));
61        id
62    }
63
64    /// Returns the original `Filter` of an `eth_newFilter`
65    pub async fn get_log_filter(&self, id: &str) -> Option<Filter> {
66        let mut filters = self.active_filters.lock().await;
67        if let Some((EthFilter::Logs(log), deadline)) = filters.get_mut(id) {
68            *deadline = self.next_deadline();
69            return log.filter.filter.clone();
70        }
71        None
72    }
73
74    /// Removes the filter identified with the `id`
75    pub async fn uninstall_filter(&self, id: &str) -> Option<EthFilter<N>> {
76        trace!(target: "node::filter", "Uninstalling filter id {}", id);
77        self.active_filters.lock().await.remove(id).map(|(f, _)| f)
78    }
79
80    /// The duration how long to keep alive stale filters
81    pub const fn keep_alive(&self) -> Duration {
82        self.keepalive
83    }
84
85    /// Returns the timestamp after which a filter should expire
86    fn next_deadline(&self) -> Instant {
87        Instant::now() + self.keep_alive()
88    }
89
90    /// Evict all filters that weren't updated and reached there deadline
91    pub async fn evict(&self) {
92        trace!(target: "node::filter", "Evicting stale filters");
93        let now = Instant::now();
94        let mut active_filters = self.active_filters.lock().await;
95        active_filters.retain(|id, (_, deadline)| {
96            if now > *deadline {
97                trace!(target: "node::filter",?id, "Evicting stale filter");
98                return false;
99            }
100            true
101        });
102    }
103}
104
105impl<N: Network> Filters<N>
106where
107    N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
108{
109    pub async fn get_filter_changes(&self, id: &str) -> ResponseResult {
110        {
111            let mut filters = self.active_filters.lock().await;
112            if let Some((filter, deadline)) = filters.get_mut(id) {
113                let resp = filter
114                    .next()
115                    .await
116                    .unwrap_or_else(|| ResponseResult::success(Vec::<()>::new()));
117                *deadline = self.next_deadline();
118                return resp;
119            }
120        }
121        warn!(target: "node::filter", "No filter found for {}", id);
122        ResponseResult::error(RpcError {
123            code: ErrorCode::ServerError(-32000),
124            message: "filter not found".into(),
125            data: None,
126        })
127    }
128}
129
130impl<N: Network> Default for Filters<N> {
131    fn default() -> Self {
132        Self {
133            active_filters: Arc::new(Default::default()),
134            keepalive: Duration::from_secs(ACTIVE_FILTER_TIMEOUT_SECS),
135        }
136    }
137}
138
139/// returns a new random hex id
140fn new_id() -> String {
141    SubscriptionId::random_hex().to_string()
142}
143
144/// Represents a poll based filter
145pub enum EthFilter<N: Network> {
146    Logs(Box<LogsFilter<N>>),
147    Blocks(ChainNotifications),
148    PendingTransactions(Receiver<TxHash>),
149    FullPendingTransactions(mpsc::Receiver<AnyRpcTransaction>),
150}
151
152impl<N: Network> std::fmt::Debug for EthFilter<N> {
153    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
154        match self {
155            Self::Logs(_) => f.debug_tuple("Logs").finish(),
156            Self::Blocks(_) => f.debug_tuple("Blocks").finish(),
157            Self::PendingTransactions(_) => f.debug_tuple("PendingTransactions").finish(),
158            Self::FullPendingTransactions(_) => f.debug_tuple("FullPendingTransactions").finish(),
159        }
160    }
161}
162
163impl<N: Network> Stream for EthFilter<N>
164where
165    N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
166{
167    type Item = ResponseResult;
168
169    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
170        let pin = self.get_mut();
171        match pin {
172            Self::Logs(logs) => Poll::Ready(Some(Ok(logs.poll(cx)).to_rpc_result())),
173            Self::Blocks(blocks) => {
174                let mut new_blocks = Vec::new();
175                while let Poll::Ready(Some(notification)) = blocks.poll_next_unpin(cx) {
176                    if let Some(block) = notification.as_new_block() {
177                        new_blocks.push(block.hash);
178                    }
179                }
180                Poll::Ready(Some(Ok(new_blocks).to_rpc_result()))
181            }
182            Self::PendingTransactions(tx) => {
183                let mut new_txs = Vec::new();
184                while let Poll::Ready(Some(tx_hash)) = tx.poll_next_unpin(cx) {
185                    new_txs.push(tx_hash);
186                }
187                Poll::Ready(Some(Ok(new_txs).to_rpc_result()))
188            }
189            Self::FullPendingTransactions(tx) => {
190                let mut new_txs = Vec::new();
191                while let Poll::Ready(Some(tx)) = tx.poll_recv(cx) {
192                    new_txs.push(tx);
193                }
194                Poll::Ready(Some(Ok(new_txs).to_rpc_result()))
195            }
196        }
197    }
198}
199
200/// Listens for new blocks and matching logs emitted in that block
201pub struct LogsFilter<N: Network> {
202    /// listener for new blocks
203    pub blocks: ChainNotifications,
204    /// accessor for block storage
205    pub storage: StorageInfo<N>,
206    /// matcher with all provided filter params
207    pub filter: FilteredParams,
208    /// existing logs that matched the filter when the listener was installed
209    ///
210    /// They'll be returned on the first poll
211    pub historic: Option<Vec<Log>>,
212}
213
214impl<N: Network> std::fmt::Debug for LogsFilter<N> {
215    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
216        f.debug_struct("LogsFilter").field("filter", &self.filter).finish_non_exhaustive()
217    }
218}
219
220impl<N: Network> LogsFilter<N>
221where
222    N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
223{
224    /// Returns all the logs since the last time this filter was polled
225    pub fn poll(&mut self, cx: &mut Context<'_>) -> Vec<Log> {
226        let mut logs = self.historic.take().unwrap_or_default();
227        while let Poll::Ready(Some(notification)) = self.blocks.poll_next_unpin(cx) {
228            match notification {
229                ChainNotification::Block(block) => {
230                    let b = self.storage.block(block.hash);
231                    let receipts = self.storage.receipts(block.hash);
232                    if let (Some(receipts), Some(block)) = (receipts, b) {
233                        logs.extend(filter_logs(block, receipts, &self.filter))
234                    }
235                }
236                ChainNotification::RemovedLogs(removed) => {
237                    logs.extend(filter_removed_logs(&removed, &self.filter))
238                }
239            }
240        }
241        logs
242    }
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248    use crate::{NodeConfig, spawn};
249
250    #[tokio::test(flavor = "multi_thread")]
251    async fn log_queries_refresh_expiry_without_preserving_idle_filters() {
252        let (api, _handle) = spawn(NodeConfig::test()).await;
253        let filters = Filters::default();
254        let mut ids = Vec::new();
255        for _ in 0..3 {
256            ids.push(
257                filters
258                    .add_filter(EthFilter::Logs(Box::new(LogsFilter {
259                        blocks: api.new_block_notifications(),
260                        storage: api.storage_info(),
261                        filter: FilteredParams::new(Some(Filter::default())),
262                        historic: None,
263                    })))
264                    .await,
265            );
266        }
267        let block_id = filters.add_filter(EthFilter::Blocks(api.new_block_notifications())).await;
268        // Expire all filters before polling, without a wall-clock sleep or eviction race.
269        let expired = Instant::now() - Duration::from_secs(1);
270        for (_, deadline) in filters.active_filters.lock().await.values_mut() {
271            *deadline = expired;
272        }
273
274        assert_eq!(filters.get_log_filter(&ids[0]).await, Some(Filter::default()));
275        filters.get_filter_changes(&ids[1]).await;
276        assert!(filters.get_log_filter(&block_id).await.is_none());
277        assert!(filters.get_log_filter("missing").await.is_none());
278        filters.evict().await;
279
280        let active = filters.active_filters.lock().await;
281        assert_eq!(active.len(), 2);
282        for id in &ids[..2] {
283            assert!(active.get(id).unwrap().1 > expired);
284        }
285        assert!(!active.contains_key(&ids[2]));
286        assert!(!active.contains_key(&block_id));
287    }
288}
anvil/filter.rs

anvil/
filter.rs
