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 filters = self.active_filters.lock().await;
67        if let Some((EthFilter::Logs(log), _)) = filters.get(id) {
68            return log.filter.filter.clone();
69        }
70        None
71    }
72
73    /// Removes the filter identified with the `id`
74    pub async fn uninstall_filter(&self, id: &str) -> Option<EthFilter<N>> {
75        trace!(target: "node::filter", "Uninstalling filter id {}", id);
76        self.active_filters.lock().await.remove(id).map(|(f, _)| f)
77    }
78
79    /// The duration how long to keep alive stale filters
80    pub const fn keep_alive(&self) -> Duration {
81        self.keepalive
82    }
83
84    /// Returns the timestamp after which a filter should expire
85    fn next_deadline(&self) -> Instant {
86        Instant::now() + self.keep_alive()
87    }
88
89    /// Evict all filters that weren't updated and reached there deadline
90    pub async fn evict(&self) {
91        trace!(target: "node::filter", "Evicting stale filters");
92        let now = Instant::now();
93        let mut active_filters = self.active_filters.lock().await;
94        active_filters.retain(|id, (_, deadline)| {
95            if now > *deadline {
96                trace!(target: "node::filter",?id, "Evicting stale filter");
97                return false;
98            }
99            true
100        });
101    }
102}
103
104impl<N: Network> Filters<N>
105where
106    N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
107{
108    pub async fn get_filter_changes(&self, id: &str) -> ResponseResult {
109        {
110            let mut filters = self.active_filters.lock().await;
111            if let Some((filter, deadline)) = filters.get_mut(id) {
112                let resp = filter
113                    .next()
114                    .await
115                    .unwrap_or_else(|| ResponseResult::success(Vec::<()>::new()));
116                *deadline = self.next_deadline();
117                return resp;
118            }
119        }
120        warn!(target: "node::filter", "No filter found for {}", id);
121        ResponseResult::error(RpcError {
122            code: ErrorCode::ServerError(-32000),
123            message: "filter not found".into(),
124            data: None,
125        })
126    }
127}
128
129impl<N: Network> Default for Filters<N> {
130    fn default() -> Self {
131        Self {
132            active_filters: Arc::new(Default::default()),
133            keepalive: Duration::from_secs(ACTIVE_FILTER_TIMEOUT_SECS),
134        }
135    }
136}
137
138/// returns a new random hex id
139fn new_id() -> String {
140    SubscriptionId::random_hex().to_string()
141}
142
143/// Represents a poll based filter
144pub enum EthFilter<N: Network> {
145    Logs(Box<LogsFilter<N>>),
146    Blocks(ChainNotifications),
147    PendingTransactions(Receiver<TxHash>),
148    FullPendingTransactions(mpsc::Receiver<AnyRpcTransaction>),
149}
150
151impl<N: Network> std::fmt::Debug for EthFilter<N> {
152    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153        match self {
154            Self::Logs(_) => f.debug_tuple("Logs").finish(),
155            Self::Blocks(_) => f.debug_tuple("Blocks").finish(),
156            Self::PendingTransactions(_) => f.debug_tuple("PendingTransactions").finish(),
157            Self::FullPendingTransactions(_) => f.debug_tuple("FullPendingTransactions").finish(),
158        }
159    }
160}
161
162impl<N: Network> Stream for EthFilter<N>
163where
164    N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
165{
166    type Item = ResponseResult;
167
168    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
169        let pin = self.get_mut();
170        match pin {
171            Self::Logs(logs) => Poll::Ready(Some(Ok(logs.poll(cx)).to_rpc_result())),
172            Self::Blocks(blocks) => {
173                let mut new_blocks = Vec::new();
174                while let Poll::Ready(Some(notification)) = blocks.poll_next_unpin(cx) {
175                    if let Some(block) = notification.as_new_block() {
176                        new_blocks.push(block.hash);
177                    }
178                }
179                Poll::Ready(Some(Ok(new_blocks).to_rpc_result()))
180            }
181            Self::PendingTransactions(tx) => {
182                let mut new_txs = Vec::new();
183                while let Poll::Ready(Some(tx_hash)) = tx.poll_next_unpin(cx) {
184                    new_txs.push(tx_hash);
185                }
186                Poll::Ready(Some(Ok(new_txs).to_rpc_result()))
187            }
188            Self::FullPendingTransactions(tx) => {
189                let mut new_txs = Vec::new();
190                while let Poll::Ready(Some(tx)) = tx.poll_recv(cx) {
191                    new_txs.push(tx);
192                }
193                Poll::Ready(Some(Ok(new_txs).to_rpc_result()))
194            }
195        }
196    }
197}
198
199/// Listens for new blocks and matching logs emitted in that block
200pub struct LogsFilter<N: Network> {
201    /// listener for new blocks
202    pub blocks: ChainNotifications,
203    /// accessor for block storage
204    pub storage: StorageInfo<N>,
205    /// matcher with all provided filter params
206    pub filter: FilteredParams,
207    /// existing logs that matched the filter when the listener was installed
208    ///
209    /// They'll be returned on the first poll
210    pub historic: Option<Vec<Log>>,
211}
212
213impl<N: Network> std::fmt::Debug for LogsFilter<N> {
214    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
215        f.debug_struct("LogsFilter").field("filter", &self.filter).finish_non_exhaustive()
216    }
217}
218
219impl<N: Network> LogsFilter<N>
220where
221    N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
222{
223    /// Returns all the logs since the last time this filter was polled
224    pub fn poll(&mut self, cx: &mut Context<'_>) -> Vec<Log> {
225        let mut logs = self.historic.take().unwrap_or_default();
226        while let Poll::Ready(Some(notification)) = self.blocks.poll_next_unpin(cx) {
227            match notification {
228                ChainNotification::Block(block) => {
229                    let b = self.storage.block(block.hash);
230                    let receipts = self.storage.receipts(block.hash);
231                    if let (Some(receipts), Some(block)) = (receipts, b) {
232                        logs.extend(filter_logs(block, receipts, &self.filter))
233                    }
234                }
235                ChainNotification::RemovedLogs(removed) => {
236                    logs.extend(filter_removed_logs(&removed, &self.filter))
237                }
238            }
239        }
240        logs
241    }
242}
anvil/filter.rs

anvil/
filter.rs
