|
1 | 1 | //! Filter kinds for subscriptions and polling filters. |
2 | 2 |
|
3 | | -use crate::interest::{NewBlockNotification, filters::FilterOutput, subs::SubscriptionBuffer}; |
| 3 | +use crate::interest::{ |
| 4 | + NewBlockNotification, ReorgNotification, filters::FilterOutput, subs::SubscriptionBuffer, |
| 5 | +}; |
4 | 6 | use alloy::rpc::types::{Filter, Header, Log}; |
5 | 7 | use std::collections::VecDeque; |
6 | 8 |
|
@@ -94,4 +96,116 @@ impl InterestKind { |
94 | 96 | Self::Block => SubscriptionBuffer::Block(VecDeque::new()), |
95 | 97 | } |
96 | 98 | } |
| 99 | + |
| 100 | + /// Filter a reorg notification for a subscription, producing a buffer of |
| 101 | + /// removed logs (with `removed: true`) that match this filter. |
| 102 | + /// |
| 103 | + /// Block subscriptions return an empty buffer — the Ethereum JSON-RPC |
| 104 | + /// spec does not push removed headers for `newHeads` subscriptions. |
| 105 | + pub(crate) fn filter_reorg_for_sub(&self, reorg: ReorgNotification) -> SubscriptionBuffer { |
| 106 | + let Some(filter) = self.as_filter() else { |
| 107 | + return self.empty_sub_buffer(); |
| 108 | + }; |
| 109 | + |
| 110 | + let logs: VecDeque<Log> = reorg |
| 111 | + .removed_blocks |
| 112 | + .into_iter() |
| 113 | + .flat_map(|block| { |
| 114 | + let block_hash = block.header.hash_slow(); |
| 115 | + let block_number = block.header.number; |
| 116 | + let block_timestamp = block.header.timestamp; |
| 117 | + block.logs.into_iter().filter(move |log| filter.matches(log)).map(move |log| Log { |
| 118 | + inner: log, |
| 119 | + block_hash: Some(block_hash), |
| 120 | + block_number: Some(block_number), |
| 121 | + block_timestamp: Some(block_timestamp), |
| 122 | + transaction_hash: None, |
| 123 | + transaction_index: None, |
| 124 | + log_index: None, |
| 125 | + removed: true, |
| 126 | + }) |
| 127 | + }) |
| 128 | + .collect(); |
| 129 | + |
| 130 | + SubscriptionBuffer::Log(logs) |
| 131 | + } |
| 132 | +} |
| 133 | + |
| 134 | +#[cfg(test)] |
| 135 | +mod tests { |
| 136 | + use super::*; |
| 137 | + use crate::interest::RemovedBlock; |
| 138 | + use alloy::primitives::{Address, B256, Bytes, LogData, address, b256}; |
| 139 | + |
| 140 | + fn test_log(addr: Address, topic: B256) -> alloy::primitives::Log { |
| 141 | + alloy::primitives::Log { |
| 142 | + address: addr, |
| 143 | + data: LogData::new_unchecked(vec![topic], Bytes::new()), |
| 144 | + } |
| 145 | + } |
| 146 | + |
| 147 | + fn test_header(number: u64) -> alloy::consensus::Header { |
| 148 | + alloy::consensus::Header { number, timestamp: 1_000_000 + number, ..Default::default() } |
| 149 | + } |
| 150 | + |
| 151 | + fn test_filter(addr: Address) -> Filter { |
| 152 | + Filter::new().address(addr) |
| 153 | + } |
| 154 | + |
| 155 | + #[test] |
| 156 | + fn filter_reorg_for_sub_matches_logs() { |
| 157 | + let addr = address!("0x0000000000000000000000000000000000000001"); |
| 158 | + let topic = b256!("0x0000000000000000000000000000000000000000000000000000000000000001"); |
| 159 | + |
| 160 | + let header = test_header(11); |
| 161 | + let kind = InterestKind::Log(Box::new(test_filter(addr))); |
| 162 | + let reorg = ReorgNotification { |
| 163 | + common_ancestor: 10, |
| 164 | + removed_blocks: vec![RemovedBlock { |
| 165 | + header: header.clone(), |
| 166 | + logs: vec![test_log(addr, topic)], |
| 167 | + }], |
| 168 | + }; |
| 169 | + |
| 170 | + let buf = kind.filter_reorg_for_sub(reorg); |
| 171 | + let SubscriptionBuffer::Log(logs) = buf else { panic!("expected Log buffer") }; |
| 172 | + |
| 173 | + assert_eq!(logs.len(), 1); |
| 174 | + assert!(logs[0].removed); |
| 175 | + assert_eq!(logs[0].inner.address, addr); |
| 176 | + assert_eq!(logs[0].block_hash.unwrap(), header.hash_slow()); |
| 177 | + assert_eq!(logs[0].block_number.unwrap(), 11); |
| 178 | + assert_eq!(logs[0].block_timestamp.unwrap(), 1_000_011); |
| 179 | + } |
| 180 | + |
| 181 | + #[test] |
| 182 | + fn filter_reorg_for_sub_filters_non_matching() { |
| 183 | + let addr = address!("0x0000000000000000000000000000000000000001"); |
| 184 | + let other = address!("0x0000000000000000000000000000000000000002"); |
| 185 | + let topic = b256!("0x0000000000000000000000000000000000000000000000000000000000000001"); |
| 186 | + |
| 187 | + let kind = InterestKind::Log(Box::new(test_filter(addr))); |
| 188 | + let reorg = ReorgNotification { |
| 189 | + common_ancestor: 10, |
| 190 | + removed_blocks: vec![RemovedBlock { |
| 191 | + header: test_header(11), |
| 192 | + logs: vec![test_log(other, topic)], |
| 193 | + }], |
| 194 | + }; |
| 195 | + |
| 196 | + let buf = kind.filter_reorg_for_sub(reorg); |
| 197 | + let SubscriptionBuffer::Log(logs) = buf else { panic!("expected Log buffer") }; |
| 198 | + assert!(logs.is_empty()); |
| 199 | + } |
| 200 | + |
| 201 | + #[test] |
| 202 | + fn filter_reorg_for_sub_block_returns_empty() { |
| 203 | + let reorg = ReorgNotification { |
| 204 | + common_ancestor: 10, |
| 205 | + removed_blocks: vec![RemovedBlock { header: test_header(11), logs: vec![] }], |
| 206 | + }; |
| 207 | + |
| 208 | + let buf = InterestKind::Block.filter_reorg_for_sub(reorg); |
| 209 | + assert!(buf.is_empty()); |
| 210 | + } |
97 | 211 | } |
0 commit comments