reth_engine_tree/tree/payload_processor/
receipt_root_task.rs

1//! Receipt root computation in a background task.
2//!
3//! This module provides a streaming receipt root builder that computes the receipt trie root
4//! in a background thread. Receipts are sent via a channel with their index, and for each
5//! receipt received, the builder incrementally flushes leaves to the underlying
6//! [`OrderedTrieRootEncodedBuilder`] when possible. When the channel closes, the task returns the
7//! computed root.
8
9use alloy_eips::Encodable2718;
10use alloy_primitives::{map::HashMap, Bloom, B256};
11use crossbeam_channel::Receiver;
12use reth_primitives_traits::Receipt;
13use reth_trie_common::ordered_root::OrderedTrieRootEncodedBuilder;
14use tokio::sync::oneshot;
15use tracing::debug_span;
16
17const RECEIPT_ENCODE_BUF_INITIAL_CAPACITY: usize = 512;
18
19/// Receipt with index, ready to be sent to the background task for encoding and trie building.
20#[derive(Debug, Clone)]
21pub struct IndexedReceipt<R> {
22    /// The transaction index within the block.
23    pub index: usize,
24    /// The receipt.
25    pub receipt: R,
26}
27
28impl<R> IndexedReceipt<R> {
29    /// Creates a new indexed receipt.
30    #[inline]
31    pub const fn new(index: usize, receipt: R) -> Self {
32        Self { index, receipt }
33    }
34}
35
36/// Handle for running the receipt root computation in a background task.
37///
38/// This struct holds the channels needed to receive receipts and send the result.
39/// Use [`Self::run`] to execute the computation (typically in a spawned blocking task).
40#[derive(Debug)]
41pub struct ReceiptRootTaskHandle<R> {
42    /// Receiver for indexed receipts.
43    receipt_rx: Receiver<IndexedReceipt<R>>,
44    /// Sender for the computed result.
45    result_tx: oneshot::Sender<(B256, Bloom)>,
46}
47
48impl<R: Receipt> ReceiptRootTaskHandle<R> {
49    /// Creates a new handle from the receipt receiver and result sender channels.
50    pub const fn new(
51        receipt_rx: Receiver<IndexedReceipt<R>>,
52        result_tx: oneshot::Sender<(B256, Bloom)>,
53    ) -> Self {
54        Self { receipt_rx, result_tx }
55    }
56
57    /// Runs the receipt root computation, consuming the handle.
58    ///
59    /// This method receives indexed receipts from the channel, encodes them,
60    /// and builds the trie incrementally. When all receipts have been received
61    /// (channel closed), it sends the result through the oneshot channel.
62    ///
63    /// This is designed to be called inside a blocking task (e.g., via
64    /// `executor.spawn_blocking(move || handle.run(receipts_len))`).
65    ///
66    /// # Arguments
67    ///
68    /// * `receipts_len` - The total number of receipts expected. This is needed to correctly order
69    ///   the trie keys according to RLP encoding rules.
70    pub fn run(self, receipts_len: impl Into<Option<usize>>) {
71        let receipts_len = receipts_len.into();
72
73        let _span = debug_span!(
74            target: "engine::tree::payload_processor",
75            "receipt_root",
76            receipts_len,
77        )
78        .entered();
79
80        let mut builder = OrderedTrieRootEncodedBuilder::new();
81        let mut aggregated_bloom = Bloom::ZERO;
82        let mut encode_buf = Vec::with_capacity(RECEIPT_ENCODE_BUF_INITIAL_CAPACITY);
83        let mut next = 0usize;
84        let mut pending = HashMap::new();
85
86        let mut push = |receipt: R| {
87            let receipt_with_bloom = receipt.with_bloom_ref();
88
89            encode_buf.clear();
90            receipt_with_bloom.encode_2718(&mut encode_buf);
91
92            aggregated_bloom |= *receipt_with_bloom.bloom_ref();
93            builder.push_next(&encode_buf);
94        };
95
96        for indexed_receipt in self.receipt_rx {
97            if indexed_receipt.index == next {
98                push(indexed_receipt.receipt);
99                next += 1;
100
101                while let Some(receipt) = pending.remove(&next) {
102                    push(receipt);
103                    next += 1;
104                }
105            } else {
106                pending.insert(indexed_receipt.index, indexed_receipt.receipt);
107            }
108        }
109
110        if receipts_len.is_some_and(|len| len != next) {
111            tracing::error!(
112                target: "engine::tree::payload_processor",
113                expected = receipts_len,
114                received = next,
115                "Receipt root task received incomplete receipts, execution likely aborted"
116            );
117            return;
118        }
119
120        if !pending.is_empty() {
121            tracing::error!(
122                target: "engine::tree::payload_processor",
123                received = next,
124                pending = pending.len(),
125                "Receipt root task received gapped receipts"
126            );
127            return;
128        }
129
130        let _ = self.result_tx.send((builder.finalize(), aggregated_bloom));
131    }
132}
133
134#[cfg(test)]
135mod tests {
136
137    use super::*;
138    use alloy_consensus::{proofs::calculate_receipt_root, TxReceipt};
139    use alloy_primitives::{b256, Address, Bytes, Log};
140    use crossbeam_channel::bounded;
141    use reth_ethereum_primitives::{Receipt, TxType};
142
143    #[tokio::test]
144    async fn test_receipt_root_task_empty() {
145        let (_tx, rx) = bounded::<IndexedReceipt<Receipt>>(1);
146        let (result_tx, result_rx) = oneshot::channel();
147        drop(_tx);
148
149        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
150        tokio::task::spawn_blocking(move || handle.run(0)).await.unwrap();
151
152        let (root, bloom) = result_rx.await.unwrap();
153
154        // Empty trie root
155        assert_eq!(root, reth_trie_common::EMPTY_ROOT_HASH);
156        assert_eq!(bloom, Bloom::ZERO);
157    }
158
159    #[tokio::test]
160    async fn test_receipt_root_task_single_receipt() {
161        let receipts: Vec<Receipt> = vec![Receipt::default()];
162
163        let (tx, rx) = bounded(1);
164        let (result_tx, result_rx) = oneshot::channel();
165        let receipts_len = receipts.len();
166
167        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
168        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
169
170        for (i, receipt) in receipts.clone().into_iter().enumerate() {
171            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
172        }
173        drop(tx);
174
175        join_handle.await.unwrap();
176        let (root, _bloom) = result_rx.await.unwrap();
177
178        // Verify against the standard calculation
179        let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
180        let expected_root = calculate_receipt_root(&receipts_with_bloom);
181
182        assert_eq!(root, expected_root);
183    }
184
185    #[tokio::test]
186    async fn test_receipt_root_task_multiple_receipts() {
187        let receipts: Vec<Receipt> = vec![Receipt::default(); 5];
188
189        let (tx, rx) = bounded(4);
190        let (result_tx, result_rx) = oneshot::channel();
191        let receipts_len = receipts.len();
192
193        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
194        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
195
196        for (i, receipt) in receipts.into_iter().enumerate() {
197            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
198        }
199        drop(tx);
200
201        join_handle.await.unwrap();
202        let (root, bloom) = result_rx.await.unwrap();
203
204        // Verify against expected values from existing test
205        assert_eq!(
206            root,
207            b256!("0x61353b4fb714dc1fccacbf7eafc4273e62f3d1eed716fe41b2a0cd2e12c63ebc")
208        );
209        assert_eq!(bloom, Bloom::ZERO);
210    }
211
212    #[tokio::test]
213    async fn test_receipt_root_matches_standard_calculation() {
214        // Create some receipts with actual data
215        let receipts = vec![
216            Receipt {
217                tx_type: TxType::Legacy,
218                cumulative_gas_used: 21000,
219                success: true,
220                logs: vec![],
221            },
222            Receipt {
223                tx_type: TxType::Eip1559,
224                cumulative_gas_used: 42000,
225                success: true,
226                logs: vec![Log {
227                    address: Address::ZERO,
228                    data: alloy_primitives::LogData::new_unchecked(vec![B256::ZERO], Bytes::new()),
229                }],
230            },
231            Receipt {
232                tx_type: TxType::Eip2930,
233                cumulative_gas_used: 63000,
234                success: false,
235                logs: vec![],
236            },
237        ];
238
239        // Calculate expected values first (before we move receipts)
240        let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
241        let expected_root = calculate_receipt_root(&receipts_with_bloom);
242        let expected_bloom =
243            receipts_with_bloom.iter().fold(Bloom::ZERO, |bloom, r| bloom | r.bloom_ref());
244
245        // Calculate using the task
246        let (tx, rx) = bounded(4);
247        let (result_tx, result_rx) = oneshot::channel();
248        let receipts_len = receipts.len();
249
250        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
251        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
252
253        for (i, receipt) in receipts.into_iter().enumerate() {
254            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
255        }
256        drop(tx);
257
258        join_handle.await.unwrap();
259        let (task_root, task_bloom) = result_rx.await.unwrap();
260
261        assert_eq!(task_root, expected_root);
262        assert_eq!(task_bloom, expected_bloom);
263    }
264
265    #[tokio::test]
266    async fn test_receipt_root_task_out_of_order() {
267        let receipts: Vec<Receipt> = vec![Receipt::default(); 5];
268
269        // Calculate expected values first (before we move receipts)
270        let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
271        let expected_root = calculate_receipt_root(&receipts_with_bloom);
272
273        let (tx, rx) = bounded(4);
274        let (result_tx, result_rx) = oneshot::channel();
275        let receipts_len = receipts.len();
276
277        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
278        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
279
280        // Send in reverse order to test out-of-order handling
281        for (i, receipt) in receipts.into_iter().enumerate().rev() {
282            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
283        }
284        drop(tx);
285
286        join_handle.await.unwrap();
287        let (root, _bloom) = result_rx.await.unwrap();
288
289        assert_eq!(root, expected_root);
290    }
291}
reth_engine_tree/tree/payload_processor/receipt_root_task.rs

reth_engine_tree/tree/payload_processor/
receipt_root_task.rs
