perf(engine): decode payload transactions once

crates/engine/primitives/src/lib.rs

@@ -16,7 +16,7 @@ use reth_payload_primitives::{
     EngineApiMessageVersion, EngineObjectValidationError, InvalidPayloadAttributesError,
     NewPayloadError, PayloadAttributes, PayloadOrAttributes, PayloadTypes,
 };
-use reth_primitives_traits::{Block, RecoveredBlock, SealedBlock, SealedHeader};
+use reth_primitives_traits::{Block, BlockBody, RecoveredBlock, SealedBlock, SealedHeader};
 use reth_storage_api::{errors::ProviderResult, StateProviderBox};
 use reth_trie_common::HashedPostState;
 use serde::{de::DeserializeOwned, Serialize};
@@ -212,6 +212,45 @@ pub trait PayloadValidator<Types: PayloadTypes>: Send + Sync + Unpin + 'static {
         sealed_block.try_recover().map_err(|e| NewPayloadError::Other(e.into()))
     }
 
+    /// Returns `true` if this validator implements
+    /// [`convert_payload_to_block_with_tx_stream`](Self::convert_payload_to_block_with_tx_stream)
+    /// and the engine should stream decoded transactions to payload conversion.
+    ///
+    /// When `false` (the default), the engine does not create the stream channel and payload
+    /// conversion decodes transactions itself via
+    /// [`convert_payload_to_block`](Self::convert_payload_to_block).
+    ///
+    /// Implementations returning `true` must also override
+    /// [`convert_payload_to_block_with_tx_stream`](Self::convert_payload_to_block_with_tx_stream);
+    /// otherwise the engine streams transactions that the default implementation discards.
+    #[cfg(feature = "std")]
+    fn supports_payload_tx_stream(&self) -> bool {
+        false
+    }
+
+    /// Converts the given payload into a sealed block, taking decoded transactions from `txs`
+    /// instead of re-decoding them from the payload.
+    ///
+    /// The engine feeds `txs` from the execution-side decode fan-out, so implementations can
+    /// assemble the block body from already-decoded transactions. If the stream disconnects
+    /// before all transactions arrive, implementations must fall back to decoding from the
+    /// payload and reproduce the exact errors of
+    /// [`convert_payload_to_block`](Self::convert_payload_to_block).
+    ///
+    /// The default drops the stream and delegates to
+    /// [`convert_payload_to_block`](Self::convert_payload_to_block), so implementations that
+    /// don't opt in via [`supports_payload_tx_stream`](Self::supports_payload_tx_stream) decode
+    /// transactions themselves.
+    #[cfg(feature = "std")]
+    fn convert_payload_to_block_with_tx_stream(
+        &self,
+        payload: Types::ExecutionData,
+        txs: PayloadTxStream<Self::Block>,
+    ) -> Result<SealedBlock<Self::Block>, NewPayloadError> {
+        drop(txs);
+        self.convert_payload_to_block(payload)
+    }
+
     /// Verifies payload post-execution w.r.t. hashed state updates.
     ///
     /// `state_updates` lazily yields the block's hashed post-state; call it only if the
@@ -261,3 +300,19 @@ pub trait PayloadValidator<Types: PayloadTypes>: Send + Sync + Unpin + 'static {
         Ok(())
     }
 }
+
+/// Item streamed from the execution-side transaction decoder to payload conversion: the
+/// transaction's original payload index plus the decoded transaction.
+///
+/// The index is required because the parallel decode fan-out may send transactions out of order.
+#[cfg(feature = "std")]
+pub type PayloadTxStreamItem<B> = (usize, <<B as Block>::Body as BlockBody>::Transaction);
+
+/// Receiving half of the decoded transaction stream consumed by
+/// [`PayloadValidator::convert_payload_to_block_with_tx_stream`].
+#[cfg(feature = "std")]
+pub type PayloadTxStream<B> = std::sync::mpsc::Receiver<PayloadTxStreamItem<B>>;
+
+/// Sending half of the decoded transaction stream, fed by the engine's decode fan-out.
+#[cfg(feature = "std")]
+pub type PayloadTxStreamSender<B> = std::sync::mpsc::SyncSender<PayloadTxStreamItem<B>>;

crates/engine/tree/src/tree/payload_processor/mod.rs

@@ -11,12 +11,14 @@ use alloy_primitives::B256;
 use crossbeam_channel::{Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
 use prewarm::PrewarmMetrics;
 use rayon::prelude::*;
+use reth_engine_primitives::PayloadTxStreamSender;
 use reth_evm::{
     block::ExecutableTxParts,
     execute::{ExecutableTxFor, WithTxEnv},
-    ConfigureEvm, ConvertTx, ExecutableTxIterator, ExecutableTxTuple, SpecFor, TxEnvFor,
+    ConfigureEvm, ConvertTx, ExecutableTxIterator, ExecutableTxTuple, RecoveredTx as _, SpecFor,
+    TxEnvFor,
 };
-use reth_primitives_traits::{FastInstant as Instant, NodePrimitives};
+use reth_primitives_traits::{BlockTy, FastInstant as Instant, NodePrimitives};
 use reth_provider::{
     BlockExecutionOutput, BlockNumReader, DatabaseProviderFactory, PruneCheckpointReader,
     StageCheckpointReader, StorageSettingsCache, TryIntoHistoricalStateProvider,
@@ -75,6 +77,9 @@ type PrewarmTxReceiver<TxEnv, Recovered> = mpsc::Receiver<(usize, RecoveredTx<Tx
 type ExecuteTxReceiver<TxEnv, Recovered, Err> =
     IndexedTxReceiver<RecoveredTx<TxEnv, Recovered>, Err>;
 type ExecuteTxSender<TxEnv, Recovered, Err> = IndexedTxSender<RecoveredTx<TxEnv, Recovered>, Err>;
+/// Sender streaming decoded transactions from the fan-out to payload conversion, so conversion
+/// can assemble the block body without re-decoding the payload.
+type BodyTxSender<Evm> = PayloadTxStreamSender<BlockTy<<Evm as ConfigureEvm>::Primitives>>;
 
 /// Entrypoint for executing the payload.
 #[derive(Debug)]
@@ -164,6 +169,7 @@ where
 {
     /// Spawns transaction conversion and cache prewarming, optionally wiring prewarm output into
     /// an externally-owned state-root task.
+    #[expect(clippy::too_many_arguments)]
     #[instrument(level = "debug", target = "engine::tree::payload_processor", skip_all)]
     pub fn spawn_with_state_root_streams<P, I: ExecutableTxIterator<Evm>>(
         &self,
@@ -173,6 +179,7 @@ where
         hint_stream: Option<StateRootHintStream>,
         hashed_update_stream: Option<StateRootUpdateStream>,
         parallel_bal_execution: bool,
+        body_tx: Option<BodyTxSender<Evm>>,
     ) -> IteratorPayloadHandle<Evm, I>
     where
         P: DatabaseProviderFactory + Clone + 'static,
@@ -183,8 +190,12 @@ where
             + TryIntoHistoricalStateProvider
             + 'static,
     {
-        let (prewarm_rx, execution_rx) =
-            self.spawn_tx_iterator(transactions, env.transaction_count, parallel_bal_execution);
+        let (prewarm_rx, execution_rx) = self.spawn_tx_iterator(
+            transactions,
+            env.transaction_count,
+            parallel_bal_execution,
+            body_tx,
+        );
         let prewarm_handle = self.spawn_caching_with(
             env,
             prewarm_rx,
@@ -230,6 +241,7 @@ where
         transactions: I,
         transaction_count: usize,
         parallel_bal_execution: bool,
+        body_tx: Option<BodyTxSender<Evm>>,
     ) -> (IteratorPrewarmTxReceiver<Evm, I>, IteratorExecuteTxReceiver<Evm, I>) {
         let (prewarm_tx, prewarm_rx) = mpsc::sync_channel(transaction_count);
         let (execute_tx, execute_rx) = crossbeam_channel::bounded(transaction_count);
@@ -246,7 +258,13 @@ where
             );
             self.executor.spawn_blocking_named("tx-iterator", move || {
                 let (transactions, convert) = transactions.into_parts();
-                convert_serial(transactions.into_iter(), &convert, &prewarm_tx, &execute_tx);
+                convert_serial(
+                    transactions.into_iter(),
+                    &convert,
+                    &prewarm_tx,
+                    &execute_tx,
+                    body_tx.as_ref(),
+                );
             });
         } else {
             // Parallel path — recover signatures in parallel on rayon, stream results
@@ -268,6 +286,7 @@ where
                             .for_each(|(idx, tx)| {
                                 let tx = tx.map(|tx| {
                                     let tx = WithTxEnv::new(tx);
+                                    send_body_tx(body_tx.as_ref(), idx, &tx);
                                     let _ = prewarm_tx.send((idx, tx.clone()));
                                     tx
                                 });
@@ -284,7 +303,13 @@ where
 
                     // Convert the first few transactions sequentially so execution can
                     // start immediately without waiting for rayon work-stealing.
-                    convert_serial(iter.by_ref().take(prefetch), &convert, &prewarm_tx, &execute_tx);
+                    convert_serial(
+                        iter.by_ref().take(prefetch),
+                        &convert,
+                        &prewarm_tx,
+                        &execute_tx,
+                        body_tx.as_ref(),
+                    );
 
                     let mut iter = iter.enumerate();
 
@@ -315,6 +340,7 @@ where
 
                             for (idx, tx) in chunk {
                                 if let Ok(tx) = &tx {
+                                    send_body_tx(body_tx.as_ref(), idx, tx);
                                     let _ = prewarm_tx.send((idx, tx.clone()));
                                 }
                                 let _ = execute_tx.send((idx, tx));
@@ -486,22 +512,43 @@ fn convert_serial<RawTx, Tx, TxEnv, InnerTx, Recovered, Err, C>(
     convert: &C,
     prewarm_tx: &mpsc::SyncSender<(usize, WithTxEnv<TxEnv, Recovered>)>,
     execute_tx: &ExecuteTxSender<TxEnv, Recovered, Err>,
+    body_tx: Option<&mpsc::SyncSender<(usize, InnerTx)>>,
 ) where
     Tx: ExecutableTxParts<TxEnv, InnerTx, Recovered = Recovered>,
     TxEnv: Clone,
+    InnerTx: Clone,
+    Recovered: reth_evm::RecoveredTx<InnerTx>,
     C: ConvertTx<RawTx, Tx = Tx, Error = Err>,
 {
     for (idx, raw_tx) in iter.enumerate() {
         let tx = convert.convert(raw_tx);
         let tx = tx.map(|tx| WithTxEnv::new(tx));
         if let Ok(tx) = &tx {
+            send_body_tx(body_tx, idx, tx);
             let _ = prewarm_tx.send((idx, tx.clone()));
         }
         let _ = execute_tx.send((idx, tx));
         trace!(target: "engine::tree::payload_processor", idx, "yielded transaction");
     }
 }
 
+/// Streams a decoded transaction to payload conversion, if it opted into the stream.
+///
+/// Send errors are ignored: a dropped receiver means conversion already finished or failed and
+/// falls back to decoding the payload itself.
+fn send_body_tx<TxEnv, InnerTx, Recovered>(
+    body_tx: Option<&mpsc::SyncSender<(usize, InnerTx)>>,
+    idx: usize,
+    tx: &WithTxEnv<TxEnv, Recovered>,
+) where
+    InnerTx: Clone,
+    Recovered: reth_evm::RecoveredTx<InnerTx>,
+{
+    if let Some(body_tx) = body_tx {
+        let _ = body_tx.send((idx, tx.tx().clone()));
+    }
+}
+
 /// Handle to all the spawned tasks.
 ///
 /// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the