reth_engine_tree/tree/payload_processor/bal/
ordered_outputs.rs

1//! Ordering adapter for speculative BAL worker results.
2
3use super::{
4    worker::{BalWorkerError, BalWorkerOutput},
5    BalExecutionError,
6};
7use crossbeam_channel::Receiver;
8
9#[derive(Debug, thiserror::Error)]
10pub(super) enum OrderedWorkerOutputError {
11    /// A worker returned an error.
12    #[error(transparent)]
13    Worker(#[from] BalWorkerError),
14    /// The result channel closed before every transaction had an output.
15    #[error("BAL worker result channel closed while waiting for ordered outputs")]
16    ResultChannelClosed,
17}
18
19impl From<OrderedWorkerOutputError> for BalExecutionError {
20    fn from(err: OrderedWorkerOutputError) -> Self {
21        match err {
22            OrderedWorkerOutputError::Worker(err) => err.into(),
23            other => Self::other(other),
24        }
25    }
26}
27
28/// Returns a blocking iterator over worker outputs in transaction order.
29///
30/// Workers may finish transactions out of order. This adapter buffers future-indexed outputs and
31/// yields each output only when every earlier transaction has been yielded. It owns no execution
32/// state and performs no canonical commit work.
33///
34/// Behavior:
35/// - each yielded item has the next transaction index, `Ok` and indexed transaction errors alike: a
36///   recovery or execution error is a per-transaction result, deferred to its slot so the consumer
37///   observes it in block order
38/// - non-indexed worker errors are forwarded immediately and exhaust the iterator
39/// - closed channels before `total` outputs yield `Err` and exhaust the iterator
40/// - out-of-bounds and duplicate indices panic because they violate the internal worker/dispatcher
41///   index invariant
42pub(super) fn ordered_worker_outputs<R>(
43    result_rx: &Receiver<Result<BalWorkerOutput<R>, BalWorkerError>>,
44    total: usize,
45) -> impl Iterator<Item = Result<BalWorkerOutput<R>, OrderedWorkerOutputError>> + '_ {
46    OrderedWorkerOutputs::new(result_rx, total)
47}
48
49struct OrderedWorkerOutputs<'a, R> {
50    result_rx: &'a Receiver<Result<BalWorkerOutput<R>, BalWorkerError>>,
51    /// Slot per transaction: an out-of-order output (or its indexed transaction error) is parked
52    /// here until every earlier transaction has been yielded.
53    pending: Vec<Option<Result<BalWorkerOutput<R>, BalWorkerError>>>,
54    /// Next transaction index to yield.
55    next: usize,
56    /// Total number of transactions in the block.
57    total: usize,
58    /// Set on a fatal error; the iterator is exhausted afterwards.
59    failed: bool,
60}
61
62impl<'a, R> OrderedWorkerOutputs<'a, R> {
63    fn new(
64        result_rx: &'a Receiver<Result<BalWorkerOutput<R>, BalWorkerError>>,
65        total: usize,
66    ) -> Self {
67        Self {
68            result_rx,
69            pending: (0..total).map(|_| None).collect(),
70            next: 0,
71            total,
72            failed: false,
73        }
74    }
75}
76
77impl<R> Iterator for OrderedWorkerOutputs<'_, R> {
78    type Item = Result<BalWorkerOutput<R>, OrderedWorkerOutputError>;
79
80    fn next(&mut self) -> Option<Self::Item> {
81        if self.failed || self.next >= self.total {
82            return None;
83        }
84
85        loop {
86            if let Some(slot) = self.pending[self.next].take() {
87                self.next += 1;
88                return Some(slot.map_err(Into::into));
89            }
90
91            let (index, slot) = match self.result_rx.recv() {
92                Ok(Ok(output)) => (output.index, Ok(output)),
93                Ok(Err(err)) => {
94                    // Transaction failures are deferred to their transaction's slot so the
95                    // consumer observes them in block order: verdict precedence must follow
96                    // block position, not worker completion order. Errors without an index
97                    // cannot be attributed to a transaction and end the iteration.
98                    let tx_index = match &err {
99                        BalWorkerError::Execution { tx_index, .. } |
100                        BalWorkerError::Transaction { tx_index, .. } => *tx_index,
101                        BalWorkerError::Setup(_) => {
102                            self.failed = true;
103                            return Some(Err(err.into()));
104                        }
105                    };
106                    (tx_index, Err(err))
107                }
108                Err(_) => {
109                    self.failed = true;
110                    return Some(Err(OrderedWorkerOutputError::ResultChannelClosed));
111                }
112            };
113
114            assert!(
115                index < self.total,
116                "BAL worker returned out-of-bounds transaction index {index}; total={}",
117                self.total
118            );
119            assert!(
120                index >= self.next && self.pending[index].is_none(),
121                "BAL worker returned duplicate transaction index {index}",
122            );
123
124            self.pending[index] = Some(slot);
125        }
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132    use crate::tree::payload_processor::bal::BalExecutionError;
133    use alloy_primitives::Address;
134
135    fn output(index: usize, result: u64) -> BalWorkerOutput<u64> {
136        BalWorkerOutput { index, signer: Address::ZERO, tx_gas_limit: 0, result }
137    }
138
139    fn expect_err_contains<R>(
140        result: Result<BalWorkerOutput<R>, OrderedWorkerOutputError>,
141        text: &str,
142    ) {
143        let Err(err) = result else {
144            panic!("expected ordered worker output error");
145        };
146        assert!(err.to_string().contains(text), "expected `{err}` to contain `{text}`");
147    }
148
149    #[test]
150    fn yields_outputs_in_transaction_order() {
151        let (tx, rx) = crossbeam_channel::unbounded();
152        tx.send(Ok(output(2, 20))).unwrap();
153        tx.send(Ok(output(0, 0))).unwrap();
154        tx.send(Ok(output(1, 10))).unwrap();
155        drop(tx);
156
157        let results = ordered_worker_outputs(&rx, 3)
158            .map(|output| output.expect("ordered output").result)
159            .collect::<Vec<_>>();
160
161        assert_eq!(results, vec![0, 10, 20]);
162    }
163
164    #[test]
165    fn forwards_worker_errors_and_then_stops() {
166        let (tx, rx) = crossbeam_channel::unbounded();
167        tx.send(Err(BalWorkerError::Setup(BalExecutionError::Execution(
168            alloy_evm::block::BlockExecutionError::msg("worker failed"),
169        ))))
170        .unwrap();
171        drop(tx);
172
173        let mut outputs = ordered_worker_outputs::<u64>(&rx, 1);
174
175        expect_err_contains(outputs.next().expect("first item"), "worker failed");
176        assert!(outputs.next().is_none());
177    }
178
179    #[test]
180    fn defers_indexed_execution_errors_to_their_slot() {
181        let (tx, rx) = crossbeam_channel::unbounded();
182        tx.send(Err(BalWorkerError::Execution {
183            tx_index: 1,
184            tx_gas_limit: 42,
185            source: alloy_evm::block::BlockExecutionError::msg("bal miss"),
186        }))
187        .unwrap();
188        tx.send(Ok(output(0, 0))).unwrap();
189        drop(tx);
190
191        let mut outputs = ordered_worker_outputs(&rx, 2);
192
193        assert_eq!(outputs.next().expect("first item").expect("first output").result, 0);
194        expect_err_contains(outputs.next().expect("second item"), "transaction 1: bal miss");
195        assert!(outputs.next().is_none());
196    }
197
198    #[test]
199    fn continues_past_indexed_execution_errors() {
200        let (tx, rx) = crossbeam_channel::unbounded();
201        tx.send(Err(BalWorkerError::Execution {
202            tx_index: 0,
203            tx_gas_limit: 42,
204            source: alloy_evm::block::BlockExecutionError::msg("bal miss"),
205        }))
206        .unwrap();
207        tx.send(Ok(output(1, 10))).unwrap();
208        drop(tx);
209
210        let mut outputs = ordered_worker_outputs(&rx, 2);
211
212        expect_err_contains(outputs.next().expect("first item"), "transaction 0: bal miss");
213        assert_eq!(outputs.next().expect("second item").expect("second output").result, 10);
214        assert!(outputs.next().is_none());
215    }
216
217    #[test]
218    fn orders_recovery_and_execution_errors_by_transaction_index() {
219        let (tx, rx) = crossbeam_channel::unbounded();
220        tx.send(Err(BalWorkerError::Execution {
221            tx_index: 2,
222            tx_gas_limit: 42,
223            source: alloy_evm::block::BlockExecutionError::msg("bal miss"),
224        }))
225        .unwrap();
226        tx.send(Err(BalWorkerError::Transaction {
227            tx_index: 1,
228            source: std::io::Error::other("sig fail").into(),
229        }))
230        .unwrap();
231        tx.send(Ok(output(0, 0))).unwrap();
232        drop(tx);
233
234        let mut outputs = ordered_worker_outputs(&rx, 3);
235        assert_eq!(outputs.next().unwrap().unwrap().result, 0);
236        expect_err_contains(outputs.next().unwrap(), "transaction 1: sig fail");
237        expect_err_contains(outputs.next().unwrap(), "transaction 2: bal miss");
238        assert!(outputs.next().is_none());
239    }
240
241    #[test]
242    fn rejects_closed_channel_before_all_outputs_arrive() {
243        let (tx, rx) = crossbeam_channel::unbounded();
244        drop(tx);
245
246        let mut outputs = ordered_worker_outputs::<u64>(&rx, 1);
247
248        expect_err_contains(outputs.next().expect("first item"), "waiting for ordered outputs");
249        assert!(outputs.next().is_none());
250    }
251
252    #[test]
253    #[should_panic(expected = "out-of-bounds transaction index")]
254    fn panics_on_out_of_bounds_index() {
255        let (tx, rx) = crossbeam_channel::unbounded();
256        tx.send(Ok(output(1, 10))).unwrap();
257        drop(tx);
258
259        let mut outputs = ordered_worker_outputs(&rx, 1);
260        let _ = outputs.next();
261    }
262
263    #[test]
264    #[should_panic(expected = "duplicate transaction index")]
265    fn panics_on_duplicate_pending_index() {
266        let (tx, rx) = crossbeam_channel::unbounded();
267        tx.send(Ok(output(1, 10))).unwrap();
268        tx.send(Ok(output(1, 11))).unwrap();
269        drop(tx);
270
271        let mut outputs = ordered_worker_outputs(&rx, 2);
272        let _ = outputs.next();
273    }
274
275    #[test]
276    #[should_panic(expected = "duplicate transaction index")]
277    fn panics_on_duplicate_already_yielded_index() {
278        let (tx, rx) = crossbeam_channel::unbounded();
279        tx.send(Ok(output(0, 0))).unwrap();
280        tx.send(Ok(output(0, 1))).unwrap();
281        drop(tx);
282
283        let mut outputs = ordered_worker_outputs(&rx, 2);
284
285        assert_eq!(outputs.next().expect("first item").expect("first output").result, 0);
286        let _ = outputs.next();
287    }
288}
reth_engine_tree/tree/payload_processor/bal/ordered_outputs.rs

reth_engine_tree/tree/payload_processor/bal/
ordered_outputs.rs
