reth_network/transactions/
announcement.rs

1//! Ordered transaction announcements used by the transaction manager and fetcher.
2
3use super::constants::SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE;
4use alloy_primitives::{map::B256Set, TxHash, B128};
5use derive_more::IntoIterator;
6use reth_eth_wire::{EthVersion, HandleMempoolData, NewPooledTransactionHashes};
7
8/// An announcement with unique hashes in the order supplied by the peer.
9///
10/// Metadata comes from the wire message. Network-specific validation is performed by the
11/// transaction manager's announcement policy.
12#[derive(Debug, IntoIterator)]
13pub struct TransactionAnnouncement {
14    #[into_iterator(owned, ref)]
15    entries: Vec<AnnouncedTransaction>,
16    version: EthVersion,
17    cell_mask: Option<B128>,
18}
19
20impl TransactionAnnouncement {
21    /// Normalizes a wire announcement, keeping the first occurrence of each hash and its metadata.
22    /// The scratch set is cleared before use and its allocation is kept for subsequent messages,
23    /// unless an oversized announcement grew it past twice the announcement soft limit.
24    ///
25    /// Returns an error if the hash, type and size arrays have different lengths.
26    pub fn from_message(
27        msg: &NewPooledTransactionHashes,
28        seen: &mut B256Set,
29    ) -> alloy_rlp::Result<Self> {
30        let (metadata, cell_mask) = match msg {
31            NewPooledTransactionHashes::Eth66(_) => (None, None),
32            NewPooledTransactionHashes::Eth68(msg) => {
33                (Some((msg.types.as_slice(), msg.sizes.as_slice())), None)
34            }
35            NewPooledTransactionHashes::Eth72(msg) => {
36                (Some((msg.types.as_slice(), msg.sizes.as_slice())), msg.cell_mask)
37            }
38        };
39        if let Some((types, sizes)) = metadata {
40            for len in [types.len(), sizes.len()] {
41                if len != msg.len() {
42                    return Err(alloy_rlp::Error::ListLengthMismatch {
43                        expected: msg.len(),
44                        got: len,
45                    })
46                }
47            }
48        }
49
50        seen.clear();
51        seen.reserve(msg.len());
52        let mut entries = Vec::with_capacity(msg.len());
53        entries.extend(msg.iter_hashes().enumerate().filter(|(_, hash)| seen.insert(**hash)).map(
54            |(index, &hash)| AnnouncedTransaction {
55                hash,
56                metadata: metadata.map(|(types, sizes)| TransactionMetadata {
57                    tx_type: types[index],
58                    size: sizes[index],
59                }),
60            },
61        ));
62        if seen.capacity() > MAX_RETAINED_SCRATCH_CAPACITY {
63            *seen = B256Set::default();
64        }
65        Ok(Self { entries, version: msg.version(), cell_mask })
66    }
67
68    /// Returns the wire message version.
69    pub const fn version(&self) -> EthVersion {
70        self.version
71    }
72
73    /// Returns the eth/72 message-level cell mask, if present.
74    pub const fn cell_mask(&self) -> Option<B128> {
75        self.cell_mask
76    }
77
78    /// Returns the number of entries.
79    pub const fn len(&self) -> usize {
80        self.entries.len()
81    }
82
83    /// Returns whether there are no entries.
84    pub const fn is_empty(&self) -> bool {
85        self.entries.is_empty()
86    }
87
88    /// Iterates over the entries in announcement order.
89    pub fn iter(&self) -> impl ExactSizeIterator<Item = &AnnouncedTransaction> {
90        self.entries.iter()
91    }
92
93    /// Retains entries satisfying the predicate without changing their order.
94    pub fn retain(&mut self, f: impl FnMut(&AnnouncedTransaction) -> bool) {
95        self.entries.retain(f);
96    }
97}
98
99impl HandleMempoolData for TransactionAnnouncement {
100    fn is_empty(&self) -> bool {
101        self.is_empty()
102    }
103
104    fn len(&self) -> usize {
105        self.len()
106    }
107
108    fn retain_by_hash(&mut self, mut f: impl FnMut(&TxHash) -> bool) {
109        self.retain(|tx| f(&tx.hash));
110    }
111}
112
113/// A transaction hash and the metadata supplied by its announcing peer.
114#[derive(Clone, Copy, Debug, PartialEq, Eq)]
115pub struct AnnouncedTransaction {
116    /// The announced transaction hash.
117    pub hash: TxHash,
118    /// Type and size for eth/68 and later; absent for eth/66.
119    pub metadata: Option<TransactionMetadata>,
120}
121
122/// Transaction metadata supplied in eth/68 and later announcements.
123#[derive(Clone, Copy, Debug, PartialEq, Eq)]
124pub struct TransactionMetadata {
125    /// The transaction type byte, interpreted by the network's announcement policy.
126    pub tx_type: u8,
127    /// The announced encoded transaction size in bytes.
128    pub size: usize,
129}
130
131/// Scratch capacity kept between messages. Only oversized announcements grow the set past this
132/// bound, and dropping it avoids pinning their allocation for the caller's lifetime.
133const MAX_RETAINED_SCRATCH_CAPACITY: usize =
134    2 * SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE;
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139    use alloy_primitives::B256;
140    use reth_eth_wire::{NewPooledTransactionHashes68, NewPooledTransactionHashes72};
141
142    #[test]
143    fn dedup_preserves_first_metadata_order_and_version() {
144        let hashes = [
145            B256::repeat_byte(3),
146            B256::repeat_byte(1),
147            B256::repeat_byte(3),
148            B256::repeat_byte(2),
149        ];
150        let types = vec![1, 2, 3, 4];
151        let sizes = vec![100, 200, 300, 400];
152        let mask = Some(B128::repeat_byte(0x11));
153        let messages = [
154            NewPooledTransactionHashes::Eth66(hashes.to_vec().into()),
155            NewPooledTransactionHashes68 {
156                hashes: hashes.to_vec(),
157                types: types.clone(),
158                sizes: sizes.clone(),
159            }
160            .into(),
161            NewPooledTransactionHashes72 { hashes: hashes.to_vec(), types, sizes, cell_mask: mask }
162                .into(),
163        ];
164        let mut seen = B256Set::default();
165        for msg in messages {
166            let mut announcement = TransactionAnnouncement::from_message(&msg, &mut seen).unwrap();
167            assert_eq!(announcement.version(), msg.version());
168            assert_eq!(
169                announcement.iter().map(|tx| tx.hash).collect::<Vec<_>>(),
170                [hashes[0], hashes[1], hashes[3]]
171            );
172            let metadata = announcement.iter().map(|tx| tx.metadata).collect::<Vec<_>>();
173            if msg.version().has_eth68_metadata() {
174                assert_eq!(
175                    metadata,
176                    [(1, 100), (2, 200), (4, 400)]
177                        .map(|(tx_type, size)| Some(TransactionMetadata { tx_type, size }))
178                );
179            } else {
180                assert_eq!(metadata, [None; 3]);
181            }
182
183            // Both hash-only and entry filters must keep the remaining metadata aligned.
184            announcement.retain_by_hash(|hash| *hash != hashes[1]);
185            announcement.retain(|tx| tx.hash != hashes[0]);
186            assert_eq!(announcement.iter().next().unwrap().hash, hashes[3]);
187            assert_eq!(
188                announcement.cell_mask(),
189                if msg.version() == EthVersion::Eth72 { mask } else { None }
190            );
191        }
192    }
193
194    #[test]
195    fn rejects_misaligned_metadata_in_locally_constructed_messages() {
196        for (hashes_len, types_len, sizes_len) in
197            [(2, 1, 2), (2, 2, 1), (1, 2, 1), (1, 1, 2), (0, 1, 1)]
198        {
199            let hashes = vec![B256::ZERO; hashes_len];
200            let types = vec![2; types_len];
201            let sizes = vec![100; sizes_len];
202            for msg in [
203                NewPooledTransactionHashes68 {
204                    hashes: hashes.clone(),
205                    types: types.clone(),
206                    sizes: sizes.clone(),
207                }
208                .into(),
209                NewPooledTransactionHashes72 { hashes, types, sizes, cell_mask: None }.into(),
210            ] {
211                assert!(
212                    TransactionAnnouncement::from_message(&msg, &mut B256Set::default()).is_err()
213                );
214            }
215        }
216    }
217
218    #[test]
219    fn scratch_is_reused_between_messages_and_dropped_when_oversized() {
220        let mut seen = B256Set::default();
221        let msg = NewPooledTransactionHashes::Eth66(vec![B256::ZERO; 64].into());
222        TransactionAnnouncement::from_message(&msg, &mut seen).unwrap();
223        let capacity = seen.capacity();
224        assert!(capacity >= 64);
225        let second = TransactionAnnouncement::from_message(&msg, &mut seen).unwrap();
226        assert_eq!(second.len(), 1);
227        assert_eq!(seen.capacity(), capacity);
228
229        // A full-size announcement keeps its allocation for later messages.
230        let full_len = SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE;
231        let full = NewPooledTransactionHashes::Eth66(vec![B256::repeat_byte(1); full_len].into());
232        assert_eq!(TransactionAnnouncement::from_message(&full, &mut seen).unwrap().len(), 1);
233        let capacity = seen.capacity();
234        assert!(capacity >= full_len && capacity <= MAX_RETAINED_SCRATCH_CAPACITY);
235        assert_eq!(TransactionAnnouncement::from_message(&msg, &mut seen).unwrap().len(), 1);
236        assert_eq!(seen.capacity(), capacity);
237
238        // An oversized announcement must not pin its allocation.
239        let oversized =
240            NewPooledTransactionHashes::Eth66(vec![B256::repeat_byte(2); 4 * full_len].into());
241        assert_eq!(TransactionAnnouncement::from_message(&oversized, &mut seen).unwrap().len(), 1);
242        assert_eq!(seen.capacity(), 0);
243    }
244}
reth_network/transactions/announcement.rs

reth_network/transactions/
announcement.rs
