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 orders_recovery_and_execution_errors_by_transaction_index() {
181        let (tx, rx) = crossbeam_channel::unbounded();
182        tx.send(Err(BalWorkerError::Execution {
183            tx_index: 2,
184            tx_gas_limit: 42,
185            source: alloy_evm::block::BlockExecutionError::msg("bal miss"),
186        }))
187        .unwrap();
188        tx.send(Err(BalWorkerError::Transaction {
189            tx_index: 1,
190            source: std::io::Error::other("sig fail").into(),
191        }))
192        .unwrap();
193        tx.send(Ok(output(0, 0))).unwrap();
194        drop(tx);
195
196        let mut outputs = ordered_worker_outputs(&rx, 3);
197        assert_eq!(outputs.next().unwrap().unwrap().result, 0);
198        expect_err_contains(outputs.next().unwrap(), "transaction 1: sig fail");
199        expect_err_contains(outputs.next().unwrap(), "transaction 2: bal miss");
200        assert!(outputs.next().is_none());
201    }
202
203    #[test]
204    fn rejects_closed_channel_before_all_outputs_arrive() {
205        let (tx, rx) = crossbeam_channel::unbounded();
206        drop(tx);
207
208        let mut outputs = ordered_worker_outputs::<u64>(&rx, 1);
209
210        expect_err_contains(outputs.next().expect("first item"), "waiting for ordered outputs");
211        assert!(outputs.next().is_none());
212    }
213
214    #[test]
215    #[should_panic(expected = "out-of-bounds transaction index")]
216    fn panics_on_out_of_bounds_index() {
217        let (tx, rx) = crossbeam_channel::unbounded();
218        tx.send(Ok(output(1, 10))).unwrap();
219        drop(tx);
220
221        let mut outputs = ordered_worker_outputs(&rx, 1);
222        let _ = outputs.next();
223    }
224
225    #[test]
226    #[should_panic(expected = "duplicate transaction index")]
227    fn panics_on_duplicate_pending_index() {
228        let (tx, rx) = crossbeam_channel::unbounded();
229        tx.send(Ok(output(1, 10))).unwrap();
230        tx.send(Ok(output(1, 11))).unwrap();
231        drop(tx);
232
233        let mut outputs = ordered_worker_outputs(&rx, 2);
234        let _ = outputs.next();
235    }
236
237    #[test]
238    #[should_panic(expected = "duplicate transaction index")]
239    fn panics_on_duplicate_already_yielded_index() {
240        let (tx, rx) = crossbeam_channel::unbounded();
241        tx.send(Ok(output(0, 0))).unwrap();
242        tx.send(Ok(output(0, 1))).unwrap();
243        drop(tx);
244
245        let mut outputs = ordered_worker_outputs(&rx, 2);
246
247        assert_eq!(outputs.next().expect("first item").expect("first output").result, 0);
248        let _ = outputs.next();
249    }
250}
reth_engine_tree/tree/payload_processor/bal/ordered_outputs.rs

reth_engine_tree/tree/payload_processor/bal/
ordered_outputs.rs
