From 72f78d60e51eccd5c42bb10b59f187e96fc49335 Mon Sep 17 00:00:00 2001 From: Hesham Date: Thu, 13 Aug 2026 14:10:05 +0200 Subject: [PATCH 1/3] perf(engine): decode payload transactions once Ref #20431 (item 3) Payload transactions were RLP-decoded twice per newPayload: once in the conversion thread and again in the execution-side tx iterator. The iterator is now the single decoder: it streams each decoded transaction over a bounded channel (capacity = tx count, sends never block) into payload conversion, which assembles the block body from the stream and returns the transactions root computed from the raw payload bytes, skipping the per-tx re-encode in the pre-execution root check. Opt-in via two defaulted PayloadValidator methods; non-Ethereum implementations keep the existing double-decode behavior unchanged. If the stream disconnects before all transactions arrive (malformed tx, aborted execution, early return), conversion falls back to decoding the retained raw bytes, reproducing the existing errors. The block hash is now validated before transaction decoding (engine-API step 1); hive engine-api showed zero delta for this reordering. Deviations from the handoff spec (.scratch/decode-once/spec.md): the sender is also dropped in ensure_ok!/early-return paths because LazyHandle::try_into_inner blocks (spec assumed scope-drop suffices); the tx-root-mismatch test is uncraftable since V1 payloads carry no root field (root is derived from the raw bytes), so root correctness is locked by a seam test instead. Claude-Session: https://claude.ai/code/session_01BaDayhJk8crEcAQFxmvQF4 --- crates/engine/primitives/src/lib.rs | 55 ++++ .../tree/src/tree/payload_processor/mod.rs | 47 +++- .../engine/tree/src/tree/payload_validator.rs | 57 +++- crates/engine/tree/src/tree/tests.rs | 188 +++++++++++++ crates/ethereum/node/src/engine.rs | 18 +- crates/ethereum/payload/src/validator.rs | 251 +++++++++++++++++- 6 files changed, 600 insertions(+), 16 deletions(-) diff --git a/crates/engine/primitives/src/lib.rs b/crates/engine/primitives/src/lib.rs index 7855b803871..af66a185c83 100644 --- a/crates/engine/primitives/src/lib.rs +++ b/crates/engine/primitives/src/lib.rs @@ -212,6 +212,44 @@ pub trait PayloadValidator: 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). + #[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). + /// + /// Returns the block together with the transactions root computed from the raw payload + /// transaction bytes; `None` means the caller recomputes it from the block body. + /// + /// The default drops the stream and delegates to + /// [`convert_payload_to_block`](Self::convert_payload_to_block), preserving today's + /// behavior for implementations that don't opt in via + /// [`supports_payload_tx_stream`](Self::supports_payload_tx_stream). + #[cfg(feature = "std")] + fn convert_payload_to_block_with_tx_stream( + &self, + payload: Types::ExecutionData, + txs: PayloadTxStream, + ) -> Result<(SealedBlock, Option), NewPayloadError> { + drop(txs); + self.convert_payload_to_block(payload).map(|block| (block, None)) + } + /// 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 +299,20 @@ pub trait PayloadValidator: 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 = + (usize, <::Body as reth_primitives_traits::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 = std::sync::mpsc::Receiver>; + +/// Sending half of the decoded transaction stream, fed by the engine's decode fan-out. +#[cfg(feature = "std")] +pub type PayloadTxStreamSender = std::sync::mpsc::SyncSender>; diff --git a/crates/engine/tree/src/tree/payload_processor/mod.rs b/crates/engine/tree/src/tree/payload_processor/mod.rs index 628155f3ca5..6f1e6a1ee70 100644 --- a/crates/engine/tree/src/tree/payload_processor/mod.rs +++ b/crates/engine/tree/src/tree/payload_processor/mod.rs @@ -14,9 +14,10 @@ use rayon::prelude::*; 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::{FastInstant as Instant, NodePrimitives, TxTy}; use reth_provider::{ BlockExecutionOutput, BlockNumReader, DatabaseProviderFactory, PruneCheckpointReader, StageCheckpointReader, StorageSettingsCache, TryIntoHistoricalStateProvider, @@ -75,6 +76,9 @@ type PrewarmTxReceiver = mpsc::Receiver<(usize, RecoveredTx = IndexedTxReceiver, Err>; type ExecuteTxSender = IndexedTxSender, 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 = mpsc::SyncSender<(usize, TxTy<::Primitives>)>; /// Entrypoint for executing the payload. #[derive(Debug)] @@ -164,6 +168,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>( &self, @@ -173,6 +178,7 @@ where hint_stream: Option, hashed_update_stream: Option, parallel_bal_execution: bool, + body_tx: Option>, ) -> IteratorPayloadHandle where P: DatabaseProviderFactory + Clone + 'static, @@ -183,8 +189,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 +240,7 @@ where transactions: I, transaction_count: usize, parallel_bal_execution: bool, + body_tx: Option>, ) -> (IteratorPrewarmTxReceiver, IteratorExecuteTxReceiver) { let (prewarm_tx, prewarm_rx) = mpsc::sync_channel(transaction_count); let (execute_tx, execute_rx) = crossbeam_channel::bounded(transaction_count); @@ -246,7 +257,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 +285,9 @@ where .for_each(|(idx, tx)| { let tx = tx.map(|tx| { let tx = WithTxEnv::new(tx); + if let Some(body_tx) = &body_tx { + let _ = body_tx.send((idx, tx.tx().clone())); + } let _ = prewarm_tx.send((idx, tx.clone())); tx }); @@ -284,7 +304,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 +341,9 @@ where for (idx, tx) in chunk { if let Ok(tx) = &tx { + if let Some(body_tx) = &body_tx { + let _ = body_tx.send((idx, tx.tx().clone())); + } let _ = prewarm_tx.send((idx, tx.clone())); } let _ = execute_tx.send((idx, tx)); @@ -486,15 +515,21 @@ fn convert_serial( convert: &C, prewarm_tx: &mpsc::SyncSender<(usize, WithTxEnv)>, execute_tx: &ExecuteTxSender, + body_tx: Option<&mpsc::SyncSender<(usize, InnerTx)>>, ) where Tx: ExecutableTxParts, TxEnv: Clone, + InnerTx: Clone, + Recovered: reth_evm::RecoveredTx, C: ConvertTx, { 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 { + if let Some(body_tx) = body_tx { + let _ = body_tx.send((idx, tx.tx().clone())); + } let _ = prewarm_tx.send((idx, tx.clone())); } let _ = execute_tx.send((idx, tx)); diff --git a/crates/engine/tree/src/tree/payload_validator.rs b/crates/engine/tree/src/tree/payload_validator.rs index 9c4dde82d67..f7fc9ac878c 100644 --- a/crates/engine/tree/src/tree/payload_validator.rs +++ b/crates/engine/tree/src/tree/payload_validator.rs @@ -128,7 +128,8 @@ use alloy_primitives::Address; use reth_chain_state::{CanonicalInMemoryState, ExecutedBlock, ExecutionTimingStats}; use reth_consensus::{ConsensusError, FullConsensus, ReceiptRootBloom}; use reth_engine_primitives::{ - ConfigureEngineEvm, ExecutableTxIterator, ExecutionPayload, InvalidBlockHook, PayloadValidator, + ConfigureEngineEvm, ExecutableTxIterator, ExecutionPayload, InvalidBlockHook, PayloadTxStream, + PayloadTxStreamSender, PayloadValidator, }; use reth_errors::{BlockExecutionError, ProviderResult}; use reth_evm::{ @@ -496,10 +497,24 @@ where } }; + // Stream decoded transactions from the execution-side decode fan-out to payload + // conversion so payloads are only decoded once. Capacity equals the transaction count so + // the fan-out never blocks on this channel; a dropped receiver just means conversion + // already finished or fell back to decoding itself. + let (mut body_tx, body_rx) = if matches!(input, BlockOrPayload::Payload(_)) && + self.validator.supports_payload_tx_stream() + { + let (tx, rx) = std::sync::mpsc::sync_channel(input.transaction_count()); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + // Spawn payload conversion and basic validation on a background thread so it runs // concurrently with the rest of the function (setup + execution). For payloads this // overlaps the cost of RLP decoding + header hashing. - let validated_block = self.spawn_convert_and_validate(&input, parent_block.clone()); + let validated_block = + self.spawn_convert_and_validate(&input, parent_block.clone(), body_rx); /// A helper macro that returns the block in case there was an error /// This macro is used for early returns before block conversion @@ -508,6 +523,10 @@ where match $expr { Ok(val) => val, Err(e) => { + // Conversion may be blocked on the tx stream, which is only fed once the + // payload processor is spawned. Drop the sender first so it falls back to + // decoding instead of deadlocking the join below. + drop(body_tx.take()); let block = validated_block.try_into_inner().expect("sole handle")?; return Err(InsertBlockError::new(block, e.into()).into()) } @@ -534,6 +553,16 @@ where if input.gas_limit() > parent_block.gas_limit().saturating_mul(MAX_EXPECTED_GAS_LIMIT_MULTIPLIER) { + // Conversion may be blocked on the tx stream, which is only fed once the payload + // processor is spawned below. Drop the sender so it falls back to decoding instead + // of deadlocking the blocking `.get()`. + if let Some(sender) = body_tx.take() { + debug!( + target: "engine::tree::payload_validator", + "dropping payload tx sender before blocking pre-execution checks" + ); + drop(sender); + } // Call `.get()` to await the pre-execution checks and exit early if they fail. if validated_block.get().is_err() { return Err(validated_block @@ -550,6 +579,8 @@ where ensure_ok!(self.state_provider_builder(parent_hash, ctx.state())) else { // this is pre-validated in the tree + // Unblock conversion before the blocking join, see `ensure_ok!`. + drop(body_tx.take()); return Err(InsertBlockError::new( validated_block.try_into_inner().expect("sole handle")?, ProviderError::HeaderNotFound(parent_hash.into()).into(), @@ -635,6 +666,7 @@ where hint_stream, hashed_update_stream, parallel_bal_execution, + body_tx.take(), )); // Create optional cache stats for detailed block logging @@ -903,11 +935,16 @@ where /// Spawns a background task to convert a [`BlockOrPayload`] into a [`SealedBlock`] and perform /// basic consensus validations on it. + /// + /// If `body_rx` is provided, payload conversion assembles the block body from transactions + /// decoded by the execution-side fan-out instead of decoding them again, see + /// [`PayloadValidator::convert_payload_to_block_with_tx_stream`]. #[expect(clippy::type_complexity)] pub fn spawn_convert_and_validate( &self, input: &BlockOrPayload, parent: SealedHeader, + body_rx: Option>, ) -> LazyHandle, InsertPayloadError>> where T: PayloadTypes>, @@ -924,11 +961,12 @@ where "convert_and_validate", ) .entered(); - let block = match input { - BlockOrPayload::Block(block) => block, - BlockOrPayload::Payload(payload) => { - validator.convert_payload_to_block(payload)? - } + let (block, tx_root) = match input { + BlockOrPayload::Block(block) => (block, None), + BlockOrPayload::Payload(payload) => match body_rx { + Some(rx) => validator.convert_payload_to_block_with_tx_stream(payload, rx)?, + None => (validator.convert_payload_to_block(payload)?, None), + }, }; if let Err(e) = consensus.validate_header(block.sealed_header()) { @@ -946,7 +984,7 @@ where drop(_enter); if let Err(e) = - consensus.validate_block_pre_execution_with_tx_root(&block, None) + consensus.validate_block_pre_execution_with_tx_root(&block, tx_root) { error!(target: "engine::tree::payload_validator", ?block, "Failed to validate block {}: {e}", block.hash()); return Err(InsertBlockError::consensus_error(e, block).into()) @@ -1350,6 +1388,7 @@ where /// /// State-root tasks are prepared before this method and can provide capabilities that /// prewarm uses for BAL-derived authoritative updates or transaction-derived hints. + #[expect(clippy::too_many_arguments)] #[instrument( level = "debug", target = "engine::tree::payload_validator", @@ -1368,6 +1407,7 @@ where hint_stream: Option, hashed_update_stream: Option, parallel_bal_execution: bool, + body_tx: Option>, ) -> Result< PayloadHandle< impl ExecutableTxFor + use, @@ -1384,6 +1424,7 @@ where hint_stream, hashed_update_stream, parallel_bal_execution, + body_tx, ); self.metrics.block_validation.spawn_payload_processor.record(start.elapsed().as_secs_f64()); diff --git a/crates/engine/tree/src/tree/tests.rs b/crates/engine/tree/src/tree/tests.rs index 53cbf1d4e5f..e1e539cf830 100644 --- a/crates/engine/tree/src/tree/tests.rs +++ b/crates/engine/tree/src/tree/tests.rs @@ -3195,3 +3195,191 @@ async fn test_on_backfill_sync_finished_opstack_retriggers_backfill_to_buffered_ async fn test_on_backfill_sync_finished_eth_retriggers_backfill_to_buffered_finalized() { assert_post_backfill_recheck_retriggers_to_buffered_target(EngineApiKind::Ethereum).await; } + +// ================================================================================================ +// PAYLOAD TX STREAM TEST SUITE +// ================================================================================================ +// +// These tests drive `validate_block_with_state` with the real Ethereum payload validator, so the +// decoded transaction stream between the execution-side tx iterator and payload conversion is +// exercised end to end, including the disconnect fallback and the sender drop in the gas-limit +// spike guard. + +/// Harness driving payload validation with [`reth_node_ethereum::EthereumEngineValidator`], which +/// opts into the payload tx stream. +struct StreamingValidatorHarness { + harness: TestHarness, + validator: BasicEngineValidator< + MockEthProvider, + reth_evm_ethereum::EthEvmConfig, + reth_node_ethereum::EthereumEngineValidator, + >, +} + +impl StreamingValidatorHarness { + fn new(chain_spec: Arc) -> Self { + let harness = TestHarness::new(chain_spec.clone()); + + // Make the genesis parent resolvable for payload validation. + let genesis = SealedHeader::seal_slow(chain_spec.genesis_header().clone()); + harness.provider.add_header(genesis.hash(), genesis.clone_header()); + + let consensus = Arc::new(EthBeaconConsensus::new(chain_spec.clone())); + let overlay_manager = harness.tree.state.tree_state.overlay_manager.clone(); + let validator = BasicEngineValidator::new( + harness.provider.clone(), + consensus, + reth_evm_ethereum::EthEvmConfig::new(chain_spec.clone()), + reth_node_ethereum::EthereumEngineValidator::new(chain_spec), + TreeConfig::default(), + Box::new(NoopInvalidBlockHook::default()), + overlay_manager, + reth_tasks::Runtime::test(), + ); + + Self { harness, validator } + } + + fn genesis_header(&self) -> SealedHeader { + self.harness.tree.canonical_in_memory_state.get_canonical_head() + } + + fn validate_payload(&mut self, payload: ExecutionData) -> ValidationOutcome { + let ctx = TreeCtx::new( + &mut self.harness.tree.state, + &self.harness.tree.canonical_in_memory_state, + ); + EngineValidator::::validate_payload(&mut self.validator, payload, ctx) + } +} + +/// Builds a V1 payload on top of `parent` whose advertised block hash is consistent with its +/// (possibly garbage) raw transaction bytes. +fn streaming_payload(parent: &SealedHeader, raw_txs: Vec, gas_limit: u64) -> ExecutionData { + let mut payload = ExecutionPayloadV1 { + parent_hash: parent.hash(), + fee_recipient: alloy_primitives::Address::ZERO, + state_root: B256::ZERO, + receipts_root: B256::ZERO, + logs_bloom: Default::default(), + prev_randao: B256::ZERO, + block_number: parent.number + 1, + gas_limit, + gas_used: 0, + timestamp: parent.timestamp + 12, + extra_data: Bytes::new(), + base_fee_per_gas: alloy_primitives::U256::ZERO, + block_hash: B256::ZERO, + transactions: raw_txs, + }; + payload.block_hash = payload.clone().into_block_raw().unwrap().header.hash_slow(); + ExecutionData { payload: payload.into(), sidecar: ExecutionPayloadSidecar::none() } +} + +/// Returns an encoded, signed transaction usable as a raw payload transaction. +fn streaming_raw_tx(nonce: u64) -> Bytes { + use alloy_consensus::SignableTransaction; + use alloy_eips::eip2718::Encodable2718; + let tx = alloy_consensus::TxLegacy { + chain_id: Some(1), + nonce, + gas_price: 7, + gas_limit: 21_000, + to: alloy_primitives::TxKind::Call(alloy_primitives::Address::ZERO), + value: alloy_primitives::U256::ZERO, + input: Default::default(), + }; + let signed: reth_ethereum_primitives::TransactionSigned = + tx.into_signed(alloy_primitives::Signature::test_signature()).into(); + signed.encoded_2718().into() +} + +/// A malformed transaction with a self-consistent block hash must report the same decode error +/// as the non-streaming path (via the disconnect fallback re-decode). +#[test] +fn test_payload_tx_stream_malformed_tx_reports_decode_error() { + reth_tracing::init_test_tracing(); + + let mut harness = StreamingValidatorHarness::new(MAINNET.clone()); + let genesis = harness.genesis_header(); + + let payload = + streaming_payload(&genesis, vec![Bytes::from_static(b"garbage")], genesis.gas_limit); + + let err = harness.validate_payload(payload).unwrap_err(); + match err { + InsertPayloadError::Payload(reth_payload_primitives::NewPayloadError::Eth( + alloy_rpc_types_engine::PayloadError::Decode(_), + )) => {} + other => panic!("expected decode error, got: {other:?}"), + } +} + +/// A payload with both a bad hash and a malformed transaction must report the hash mismatch: +/// the streaming conversion validates the block hash before touching transactions. +#[test] +fn test_payload_tx_stream_block_hash_takes_precedence() { + reth_tracing::init_test_tracing(); + + let mut harness = StreamingValidatorHarness::new(MAINNET.clone()); + let genesis = harness.genesis_header(); + + let ExecutionData { payload, sidecar } = + streaming_payload(&genesis, vec![Bytes::from_static(b"garbage")], genesis.gas_limit); + let mut payload = payload.into_v1(); + payload.block_hash = B256::repeat_byte(0xab); + let payload = ExecutionData { payload: payload.into(), sidecar }; + + let err = harness.validate_payload(payload).unwrap_err(); + match err { + InsertPayloadError::Payload(reth_payload_primitives::NewPayloadError::Eth( + alloy_rpc_types_engine::PayloadError::BlockHash { .. }, + )) => {} + other => panic!("expected block hash mismatch, got: {other:?}"), + } +} + +/// A gas-limit spike blocks on pre-execution checks before the payload processor (and thus the +/// tx stream producer) is spawned. The sender must be dropped first so conversion falls back to +/// decoding instead of deadlocking. +#[test] +fn test_payload_tx_stream_gas_limit_spike_does_not_deadlock() { + reth_tracing::init_test_tracing(); + + let mut harness = StreamingValidatorHarness::new(MAINNET.clone()); + let genesis = harness.genesis_header(); + + // More than MAX_EXPECTED_GAS_LIMIT_MULTIPLIER (2x) the parent gas limit, with transactions + // so conversion actually waits on the stream. + let payload = streaming_payload(&genesis, vec![streaming_raw_tx(0)], genesis.gas_limit * 3); + + // The gas limit jump is consensus-invalid. A block-level consensus error proves conversion + // completed via the disconnect fallback (transactions decoded, block assembled) instead of + // deadlocking on the blocking pre-execution checks or failing to decode. + let err = harness.validate_payload(payload).unwrap_err(); + assert!( + matches!(err, InsertPayloadError::Block(_)), + "expected post-conversion consensus error, got: {err:?}" + ); +} + +/// Execution of the payload fails (mock EVM), but conversion must still complete via the +/// streamed transactions (or the disconnect fallback) instead of hanging. +#[test] +fn test_payload_tx_stream_aborted_execution_does_not_hang() { + reth_tracing::init_test_tracing(); + + let mut harness = StreamingValidatorHarness::new(MAINNET.clone()); + let genesis = harness.genesis_header(); + + let payload = streaming_payload(&genesis, vec![streaming_raw_tx(0)], genesis.gas_limit); + + // The crafted payload cannot pass consensus/execution. A block-level error proves the + // streamed transactions were assembled into a block (conversion terminated) rather than + // validation waiting on the tx stream forever or failing to decode. + let err = harness.validate_payload(payload).unwrap_err(); + assert!( + matches!(err, InsertPayloadError::Block(_)), + "expected post-conversion error, got: {err:?}" + ); +} diff --git a/crates/ethereum/node/src/engine.rs b/crates/ethereum/node/src/engine.rs index f1b880ab253..43c1df0fc25 100644 --- a/crates/ethereum/node/src/engine.rs +++ b/crates/ethereum/node/src/engine.rs @@ -1,12 +1,13 @@ //! Validates execution payload wrt Ethereum Execution Engine API version. +use alloy_primitives::B256; use alloy_rpc_types_engine::ExecutionData; pub use alloy_rpc_types_engine::{ ExecutionPayloadEnvelopeV2, ExecutionPayloadEnvelopeV3, ExecutionPayloadEnvelopeV4, ExecutionPayloadV1, PayloadAttributes as EthPayloadAttributes, }; use reth_chainspec::{EthChainSpec, EthereumHardforks}; -use reth_engine_primitives::{EngineApiValidator, PayloadValidator}; +use reth_engine_primitives::{EngineApiValidator, PayloadTxStream, PayloadValidator}; use reth_ethereum_payload_builder::EthereumExecutionPayloadValidator; use reth_ethereum_primitives::Block; use reth_node_api::PayloadTypes; @@ -49,6 +50,21 @@ where ) -> Result, NewPayloadError> { self.inner.ensure_well_formed_payload(payload).map_err(Into::into) } + + fn supports_payload_tx_stream(&self) -> bool { + true + } + + fn convert_payload_to_block_with_tx_stream( + &self, + payload: ExecutionData, + txs: PayloadTxStream, + ) -> Result<(SealedBlock, Option), NewPayloadError> { + self.inner + .ensure_well_formed_payload_with_tx_stream(payload, txs) + .map(|(block, tx_root)| (block, Some(tx_root))) + .map_err(Into::into) + } } impl EngineApiValidator for EthereumEngineValidator diff --git a/crates/ethereum/payload/src/validator.rs b/crates/ethereum/payload/src/validator.rs index ccace26ef80..3ed31af1aad 100644 --- a/crates/ethereum/payload/src/validator.rs +++ b/crates/ethereum/payload/src/validator.rs @@ -1,11 +1,13 @@ //! Validates execution payload wrt Ethereum consensus rules use alloy_consensus::Block; +use alloy_primitives::{Bytes, B256}; use alloy_rpc_types_engine::{ExecutionData, PayloadError}; use reth_chainspec::EthereumHardforks; use reth_payload_validator::{cancun, prague, shanghai}; use reth_primitives_traits::{Block as _, SealedBlock, SignedTransaction}; -use std::sync::Arc; +use std::sync::{mpsc, Arc}; +use tracing::{debug, debug_span}; /// Execution payload validator. #[derive(Clone, Debug)] @@ -38,6 +40,18 @@ impl EthereumExecutionPayloadValidator ) -> Result>, PayloadError> { ensure_well_formed_payload(&self.chain_spec, payload) } + + /// Streaming variant of [`Self::ensure_well_formed_payload`] that takes decoded transactions + /// from `txs` instead of decoding them from the payload. + /// + /// See also [`ensure_well_formed_payload_with_tx_stream`] + pub fn ensure_well_formed_payload_with_tx_stream( + &self, + payload: ExecutionData, + txs: mpsc::Receiver<(usize, T)>, + ) -> Result<(SealedBlock>, B256), PayloadError> { + ensure_well_formed_payload_with_tx_stream(&self.chain_spec, payload, txs) + } } /// Ensures that the given payload does not violate any consensus rules that concern the block's @@ -105,3 +119,238 @@ where Ok(sealed_block) } + +/// Streaming variant of [`ensure_well_formed_payload`] that assembles the block body from +/// transactions decoded by the engine's execution-side fan-out instead of decoding them again. +/// +/// The block hash is validated before any transaction handling, so a payload with both a bad hash +/// and a malformed transaction reports [`PayloadError::BlockHash`]. The Engine API lists the block +/// hash check first; the non-streaming path reports the decode error instead because it must +/// decode transactions to compute the hash input. +/// +/// If `txs` disconnects before all transactions arrive (e.g. a malformed transaction stopped the +/// decoder, or execution was aborted), this falls back to decoding the retained raw transaction +/// bytes, reproducing the errors of [`ensure_well_formed_payload`]. +/// +/// Returns the sealed block together with the transactions root computed from the raw payload +/// bytes, so callers can skip re-encoding transactions for root validation. +pub fn ensure_well_formed_payload_with_tx_stream( + chain_spec: ChainSpec, + payload: ExecutionData, + txs: mpsc::Receiver<(usize, T)>, +) -> Result<(SealedBlock>, B256), PayloadError> +where + ChainSpec: EthereumHardforks, + T: SignedTransaction, +{ + let ExecutionData { payload, sidecar } = payload; + + let expected_hash = payload.block_hash(); + + // Build the block with raw transaction bytes; the transactions root in the header is + // computed directly from the raw bytes, so transactions are never re-encoded. + let raw_block = payload.into_block_with_sidecar_raw(&sidecar)?; + let transactions_root = raw_block.header.transactions_root; + + // Ensure the hash included in the payload matches the block hash before waiting on any + // transactions, so header validation overlaps with the execution-side decode. + let hash = raw_block.header.hash_slow(); + if expected_hash != hash { + return Err(PayloadError::BlockHash { execution: hash, consensus: expected_hash }) + } + + let Block { header, body: raw_body } = raw_block; + + let transactions = { + let _span = debug_span!( + target: "engine::tree::payload_validator", + "assemble_body_from_stream", + ) + .entered(); + let transaction_count = raw_body.transactions.len(); + let mut slots: Vec> = Vec::new(); + slots.resize_with(transaction_count, || None); + let mut received = 0usize; + + loop { + if received == transaction_count { + break slots.into_iter().flatten().collect() + } + match txs.recv() { + Ok((idx, tx)) => { + // The parallel fan-out sends out of order, hence index-addressed slots. + if let Some(slot) = slots.get_mut(idx) && + slot.replace(tx).is_none() + { + received += 1; + } + } + Err(_) => { + // The decoder died before all transactions arrived (malformed transaction, + // aborted execution, or an engine early-return). Decode from the retained + // raw bytes to reproduce the non-streaming behavior. + debug!( + target: "engine::tree::payload_validator", + received, + transaction_count, + "payload tx stream disconnected, falling back to full decode" + ); + break decode_transactions(&raw_body.transactions)? + } + } + } + }; + + let block = Block { + header, + body: alloy_consensus::BlockBody { + transactions, + ommers: vec![], + withdrawals: raw_body.withdrawals, + }, + }; + // The header hash was already computed and verified above, no need to reseal. + let sealed_block = SealedBlock::new_unchecked(block, hash); + + shanghai::ensure_well_formed_fields( + sealed_block.body(), + chain_spec.is_shanghai_active_at_timestamp(sealed_block.timestamp), + )?; + + cancun::ensure_well_formed_fields( + &sealed_block, + sidecar.cancun(), + chain_spec.is_cancun_active_at_timestamp(sealed_block.timestamp), + )?; + + prague::ensure_well_formed_fields( + sealed_block.body(), + sidecar.prague(), + chain_spec.is_prague_active_at_timestamp(sealed_block.timestamp), + )?; + + Ok((sealed_block, transactions_root)) +} + +/// Decodes raw payload transactions, mirroring the errors produced by +/// [`ExecutionPayload::try_into_block_with_sidecar`](alloy_rpc_types_engine::ExecutionPayload::try_into_block_with_sidecar). +fn decode_transactions(raw: &[Bytes]) -> Result, PayloadError> { + raw.iter() + .map(|tx| { + T::decode_2718_exact(tx.as_ref()) + .map_err(alloy_rlp::Error::from) + .map_err(PayloadError::from) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use alloy_consensus::{BlockBody, SignableTransaction, TxLegacy}; + use alloy_primitives::{Address, Signature, TxKind, B256, U256}; + use alloy_rpc_types_engine::{ExecutionPayloadSidecar, ExecutionPayloadV1}; + use reth_chainspec::MAINNET; + use reth_ethereum_primitives::TransactionSigned; + + fn signed_tx(nonce: u64) -> TransactionSigned { + let tx = TxLegacy { + chain_id: Some(1), + nonce, + gas_price: 7, + gas_limit: 21_000, + to: TxKind::Call(Address::ZERO), + value: U256::ZERO, + input: Default::default(), + }; + tx.into_signed(Signature::test_signature()).into() + } + + /// Builds a payload whose advertised block hash is consistent with its (possibly garbage) + /// raw transaction bytes. + fn payload_with_raw_txs(raw_txs: Vec) -> ExecutionData { + let block: Block = + Block { header: Default::default(), body: BlockBody::default() }; + let mut payload = ExecutionPayloadV1::from_block_unchecked(B256::ZERO, &block); + payload.transactions = raw_txs; + payload.block_hash = payload.clone().into_block_raw().unwrap().header.hash_slow(); + ExecutionData { payload: payload.into(), sidecar: ExecutionPayloadSidecar::none() } + } + + fn payload_with_txs(txs: &[TransactionSigned]) -> ExecutionData { + use alloy_eips::eip2718::Encodable2718; + payload_with_raw_txs(txs.iter().map(|tx| tx.encoded_2718().into()).collect()) + } + + #[test] + fn assembles_body_from_out_of_order_stream() { + let txs: Vec<_> = (0..3).map(signed_tx).collect(); + let payload = payload_with_txs(&txs); + + let expected: SealedBlock> = + ensure_well_formed_payload(&*MAINNET, payload.clone()).unwrap(); + + let (tx, rx) = mpsc::sync_channel(txs.len()); + for idx in [2, 0, 1] { + tx.send((idx, txs[idx].clone())).unwrap(); + } + drop(tx); + + let (block, tx_root) = + ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap(); + assert_eq!(block, expected); + assert_eq!(tx_root, expected.transactions_root); + } + + #[test] + fn falls_back_to_decoding_on_disconnect() { + let txs: Vec<_> = (0..3).map(signed_tx).collect(); + let payload = payload_with_txs(&txs); + + let expected: SealedBlock> = + ensure_well_formed_payload(&*MAINNET, payload.clone()).unwrap(); + + // Only one of three transactions arrives before the stream dies. + let (tx, rx) = mpsc::sync_channel(txs.len()); + tx.send((0, txs[0].clone())).unwrap(); + drop(tx); + + let (block, _) = ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap(); + assert_eq!(block, expected); + } + + #[test] + fn fallback_reproduces_decode_error() { + // Garbage transaction bytes but a self-consistent block hash: the early hash check + // passes and the fallback must report the same decode error as the non-streaming path. + let payload = payload_with_raw_txs(vec![Bytes::from_static(b"garbage")]); + + let expected_err = + ensure_well_formed_payload::<_, TransactionSigned>(&*MAINNET, payload.clone()) + .unwrap_err(); + + let (tx, rx) = mpsc::sync_channel::<(usize, TransactionSigned)>(1); + drop(tx); + + let err = ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap_err(); + assert_eq!(format!("{err:?}"), format!("{expected_err:?}")); + assert!(matches!(err, PayloadError::Decode(_))); + } + + #[test] + fn block_hash_checked_before_transactions() { + // Both a bad hash and a malformed transaction: the streaming path reports the hash + // mismatch because it validates the hash before touching transactions. + let ExecutionData { payload, sidecar } = + payload_with_raw_txs(vec![Bytes::from_static(b"garbage")]); + let mut payload = payload.into_v1(); + payload.block_hash = B256::repeat_byte(0xff); + let payload = ExecutionData { payload: payload.into(), sidecar }; + + let (tx, rx) = mpsc::sync_channel::<(usize, TransactionSigned)>(1); + drop(tx); + + let err = ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap_err(); + assert!(matches!(err, PayloadError::BlockHash { .. })); + } +} From ac7948819c42e4fc25a1eafe1fc7f60020b44876 Mon Sep 17 00:00:00 2001 From: Hesham Date: Fri, 14 Aug 2026 09:02:42 +0200 Subject: [PATCH 2/3] refactor(engine): simplify payload tx stream plumbing Encode the drop-sender-before-join invariant in a PendingValidatedBlock wrapper instead of three manual drops, define the body tx sender alias via the canonical PayloadTxStreamSender, and deduplicate the fork-field checks between the streaming and non-streaming payload validators. Claude-Session: https://claude.ai/code/session_01H8bWdbHjv1fuUGgyR3TZcT --- crates/engine/primitives/src/lib.rs | 4 + .../tree/src/tree/payload_processor/mod.rs | 5 +- .../engine/tree/src/tree/payload_validator.rs | 80 ++++++++++--------- crates/engine/tree/src/tree/tests.rs | 45 +++++------ crates/ethereum/payload/src/validator.rs | 43 +++++----- 5 files changed, 91 insertions(+), 86 deletions(-) diff --git a/crates/engine/primitives/src/lib.rs b/crates/engine/primitives/src/lib.rs index af66a185c83..5402b9f24a1 100644 --- a/crates/engine/primitives/src/lib.rs +++ b/crates/engine/primitives/src/lib.rs @@ -219,6 +219,10 @@ pub trait PayloadValidator: Send + Sync + Unpin + 'static { /// 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 diff --git a/crates/engine/tree/src/tree/payload_processor/mod.rs b/crates/engine/tree/src/tree/payload_processor/mod.rs index 6f1e6a1ee70..353f5fb3ab7 100644 --- a/crates/engine/tree/src/tree/payload_processor/mod.rs +++ b/crates/engine/tree/src/tree/payload_processor/mod.rs @@ -11,13 +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, RecoveredTx as _, SpecFor, TxEnvFor, }; -use reth_primitives_traits::{FastInstant as Instant, NodePrimitives, TxTy}; +use reth_primitives_traits::{BlockTy, FastInstant as Instant, NodePrimitives}; use reth_provider::{ BlockExecutionOutput, BlockNumReader, DatabaseProviderFactory, PruneCheckpointReader, StageCheckpointReader, StorageSettingsCache, TryIntoHistoricalStateProvider, @@ -78,7 +79,7 @@ type ExecuteTxReceiver = type ExecuteTxSender = IndexedTxSender, 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 = mpsc::SyncSender<(usize, TxTy<::Primitives>)>; +type BodyTxSender = PayloadTxStreamSender::Primitives>>; /// Entrypoint for executing the payload. #[derive(Debug)] diff --git a/crates/engine/tree/src/tree/payload_validator.rs b/crates/engine/tree/src/tree/payload_validator.rs index f7fc9ac878c..6c2105b6d51 100644 --- a/crates/engine/tree/src/tree/payload_validator.rs +++ b/crates/engine/tree/src/tree/payload_validator.rs @@ -143,8 +143,8 @@ use reth_payload_primitives::{ PayloadTypes, }; use reth_primitives_traits::{ - AlloyBlockHeader, BlockBody, BlockTy, FastInstant as Instant, GotExpected, NodePrimitives, - RecoveredBlock, SealedBlock, SealedHeader, SignerRecoverable, + AlloyBlockHeader, Block, BlockBody, BlockTy, FastInstant as Instant, GotExpected, + NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader, SignerRecoverable, }; use reth_provider::{ BlockExecutionOutput, BlockReader, ChangeSetReader, DatabaseProviderFactory, @@ -501,20 +501,18 @@ where // conversion so payloads are only decoded once. Capacity equals the transaction count so // the fan-out never blocks on this channel; a dropped receiver just means conversion // already finished or fell back to decoding itself. - let (mut body_tx, body_rx) = if matches!(input, BlockOrPayload::Payload(_)) && - self.validator.supports_payload_tx_stream() - { - let (tx, rx) = std::sync::mpsc::sync_channel(input.transaction_count()); - (Some(tx), Some(rx)) - } else { - (None, None) - }; + let (body_tx, body_rx) = (matches!(input, BlockOrPayload::Payload(_)) && + self.validator.supports_payload_tx_stream()) + .then(|| std::sync::mpsc::sync_channel(input.transaction_count())) + .unzip(); // Spawn payload conversion and basic validation on a background thread so it runs // concurrently with the rest of the function (setup + execution). For payloads this // overlaps the cost of RLP decoding + header hashing. - let validated_block = - self.spawn_convert_and_validate(&input, parent_block.clone(), body_rx); + let mut validated_block = PendingValidatedBlock { + handle: self.spawn_convert_and_validate(&input, parent_block.clone(), body_rx), + body_tx, + }; /// A helper macro that returns the block in case there was an error /// This macro is used for early returns before block conversion @@ -523,11 +521,7 @@ where match $expr { Ok(val) => val, Err(e) => { - // Conversion may be blocked on the tx stream, which is only fed once the - // payload processor is spawned. Drop the sender first so it falls back to - // decoding instead of deadlocking the join below. - drop(body_tx.take()); - let block = validated_block.try_into_inner().expect("sole handle")?; + let block = validated_block.into_inner()?; return Err(InsertBlockError::new(block, e.into()).into()) } } @@ -553,22 +547,9 @@ where if input.gas_limit() > parent_block.gas_limit().saturating_mul(MAX_EXPECTED_GAS_LIMIT_MULTIPLIER) { - // Conversion may be blocked on the tx stream, which is only fed once the payload - // processor is spawned below. Drop the sender so it falls back to decoding instead - // of deadlocking the blocking `.get()`. - if let Some(sender) = body_tx.take() { - debug!( - target: "engine::tree::payload_validator", - "dropping payload tx sender before blocking pre-execution checks" - ); - drop(sender); - } // Call `.get()` to await the pre-execution checks and exit early if they fail. if validated_block.get().is_err() { - return Err(validated_block - .try_into_inner() - .expect("sole handle") - .expect_err("Err result checked")) + return Err(validated_block.into_inner().expect_err("Err result checked")) } } @@ -579,10 +560,8 @@ where ensure_ok!(self.state_provider_builder(parent_hash, ctx.state())) else { // this is pre-validated in the tree - // Unblock conversion before the blocking join, see `ensure_ok!`. - drop(body_tx.take()); return Err(InsertBlockError::new( - validated_block.try_into_inner().expect("sole handle")?, + validated_block.into_inner()?, ProviderError::HeaderNotFound(parent_hash.into()).into(), ) .into()) @@ -666,7 +645,7 @@ where hint_stream, hashed_update_stream, parallel_bal_execution, - body_tx.take(), + validated_block.take_sender(), )); // Create optional cache stats for detailed block logging @@ -795,7 +774,7 @@ where } }); - let block = validated_block.try_into_inner().expect("sole handle")?; + let block = validated_block.into_inner()?; let block = block.with_senders(senders); // Wait for the receipt root computation to complete. @@ -2110,3 +2089,32 @@ impl BlockOrPayload { } } } + +/// Pending payload conversion paired with the sender feeding its transaction stream. +/// +/// Conversion may block on the stream, which is only fed once the payload processor is spawned, +/// so every join drops the sender first: conversion then falls back to decoding the payload +/// itself instead of deadlocking. +struct PendingValidatedBlock { + handle: LazyHandle, InsertPayloadError>>, + body_tx: Option>, +} + +impl PendingValidatedBlock { + /// Takes the sender for the payload processor's decode fan-out. + const fn take_sender(&mut self) -> Option> { + self.body_tx.take() + } + + /// Blocks until conversion completes and returns a reference to the result. + fn get(&mut self) -> &Result, InsertPayloadError> { + drop(self.body_tx.take()); + self.handle.get() + } + + /// Consumes the handle and returns the conversion result. + fn into_inner(mut self) -> Result, InsertPayloadError> { + drop(self.body_tx.take()); + self.handle.try_into_inner().expect("sole handle") + } +} diff --git a/crates/engine/tree/src/tree/tests.rs b/crates/engine/tree/src/tree/tests.rs index e1e539cf830..3bb7f093e8f 100644 --- a/crates/engine/tree/src/tree/tests.rs +++ b/crates/engine/tree/src/tree/tests.rs @@ -9,15 +9,16 @@ use crate::{ }; use reth_storage_overlay::OverlayManager; -use alloy_eips::eip1898::BlockWithParent; +use alloy_consensus::{SignableTransaction, TxLegacy}; +use alloy_eips::{eip1898::BlockWithParent, eip2718::Encodable2718}; use alloy_primitives::{ map::{B256Map, B256Set}, - Bytes, B256, + Address, Bytes, Signature, TxKind, B256, U256, }; use alloy_rlp::Decodable; use alloy_rpc_types_engine::{ ExecutionData, ExecutionPayloadSidecar, ExecutionPayloadV1, ForkchoiceState, - ForkchoiceUpdateError, + ForkchoiceUpdateError, PayloadError, }; use assert_matches::assert_matches; use reth_chain_state::{test_utils::TestBlockBuilder, BlockState}; @@ -25,8 +26,9 @@ use reth_chainspec::{ChainSpec, HOLESKY, MAINNET}; use reth_engine_primitives::{EngineApiValidator, ForkchoiceStatus, NoopInvalidBlockHook}; use reth_ethereum_consensus::EthBeaconConsensus; use reth_ethereum_engine_primitives::{EthEngineTypes, EthPayloadAttributes}; -use reth_ethereum_primitives::{Block, EthPrimitives}; -use reth_evm_ethereum::MockEvmConfig; +use reth_ethereum_primitives::{Block, EthPrimitives, TransactionSigned}; +use reth_evm_ethereum::{EthEvmConfig, MockEvmConfig}; +use reth_node_ethereum::EthereumEngineValidator; use reth_payload_builder::PayloadServiceCommand; use reth_primitives_traits::Block as _; use reth_provider::{test_utils::MockEthProvider, BalStoreHandle, InMemoryBalStore, RawBal}; @@ -3205,15 +3207,11 @@ async fn test_on_backfill_sync_finished_eth_retriggers_backfill_to_buffered_fina // exercised end to end, including the disconnect fallback and the sender drop in the gas-limit // spike guard. -/// Harness driving payload validation with [`reth_node_ethereum::EthereumEngineValidator`], which -/// opts into the payload tx stream. +/// Harness driving payload validation with [`EthereumEngineValidator`], which opts into the +/// payload tx stream. struct StreamingValidatorHarness { harness: TestHarness, - validator: BasicEngineValidator< - MockEthProvider, - reth_evm_ethereum::EthEvmConfig, - reth_node_ethereum::EthereumEngineValidator, - >, + validator: BasicEngineValidator, } impl StreamingValidatorHarness { @@ -3229,8 +3227,8 @@ impl StreamingValidatorHarness { let validator = BasicEngineValidator::new( harness.provider.clone(), consensus, - reth_evm_ethereum::EthEvmConfig::new(chain_spec.clone()), - reth_node_ethereum::EthereumEngineValidator::new(chain_spec), + EthEvmConfig::new(chain_spec.clone()), + EthereumEngineValidator::new(chain_spec), TreeConfig::default(), Box::new(NoopInvalidBlockHook::default()), overlay_manager, @@ -3258,7 +3256,7 @@ impl StreamingValidatorHarness { fn streaming_payload(parent: &SealedHeader, raw_txs: Vec, gas_limit: u64) -> ExecutionData { let mut payload = ExecutionPayloadV1 { parent_hash: parent.hash(), - fee_recipient: alloy_primitives::Address::ZERO, + fee_recipient: Address::ZERO, state_root: B256::ZERO, receipts_root: B256::ZERO, logs_bloom: Default::default(), @@ -3268,7 +3266,7 @@ fn streaming_payload(parent: &SealedHeader, raw_txs: Vec, gas_limit: u64) gas_used: 0, timestamp: parent.timestamp + 12, extra_data: Bytes::new(), - base_fee_per_gas: alloy_primitives::U256::ZERO, + base_fee_per_gas: U256::ZERO, block_hash: B256::ZERO, transactions: raw_txs, }; @@ -3278,19 +3276,16 @@ fn streaming_payload(parent: &SealedHeader, raw_txs: Vec, gas_limit: u64) /// Returns an encoded, signed transaction usable as a raw payload transaction. fn streaming_raw_tx(nonce: u64) -> Bytes { - use alloy_consensus::SignableTransaction; - use alloy_eips::eip2718::Encodable2718; - let tx = alloy_consensus::TxLegacy { + let tx = TxLegacy { chain_id: Some(1), nonce, gas_price: 7, gas_limit: 21_000, - to: alloy_primitives::TxKind::Call(alloy_primitives::Address::ZERO), - value: alloy_primitives::U256::ZERO, + to: TxKind::Call(Address::ZERO), + value: U256::ZERO, input: Default::default(), }; - let signed: reth_ethereum_primitives::TransactionSigned = - tx.into_signed(alloy_primitives::Signature::test_signature()).into(); + let signed: TransactionSigned = tx.into_signed(Signature::test_signature()).into(); signed.encoded_2718().into() } @@ -3309,7 +3304,7 @@ fn test_payload_tx_stream_malformed_tx_reports_decode_error() { let err = harness.validate_payload(payload).unwrap_err(); match err { InsertPayloadError::Payload(reth_payload_primitives::NewPayloadError::Eth( - alloy_rpc_types_engine::PayloadError::Decode(_), + PayloadError::Decode(_), )) => {} other => panic!("expected decode error, got: {other:?}"), } @@ -3333,7 +3328,7 @@ fn test_payload_tx_stream_block_hash_takes_precedence() { let err = harness.validate_payload(payload).unwrap_err(); match err { InsertPayloadError::Payload(reth_payload_primitives::NewPayloadError::Eth( - alloy_rpc_types_engine::PayloadError::BlockHash { .. }, + PayloadError::BlockHash { .. }, )) => {} other => panic!("expected block hash mismatch, got: {other:?}"), } diff --git a/crates/ethereum/payload/src/validator.rs b/crates/ethereum/payload/src/validator.rs index 3ed31af1aad..fd50d5d19fb 100644 --- a/crates/ethereum/payload/src/validator.rs +++ b/crates/ethereum/payload/src/validator.rs @@ -2,7 +2,7 @@ use alloy_consensus::Block; use alloy_primitives::{Bytes, B256}; -use alloy_rpc_types_engine::{ExecutionData, PayloadError}; +use alloy_rpc_types_engine::{ExecutionData, ExecutionPayloadSidecar, PayloadError}; use reth_chainspec::EthereumHardforks; use reth_payload_validator::{cancun, prague, shanghai}; use reth_primitives_traits::{Block as _, SealedBlock, SignedTransaction}; @@ -100,22 +100,7 @@ where }) } - shanghai::ensure_well_formed_fields( - sealed_block.body(), - chain_spec.is_shanghai_active_at_timestamp(sealed_block.timestamp), - )?; - - cancun::ensure_well_formed_fields( - &sealed_block, - sidecar.cancun(), - chain_spec.is_cancun_active_at_timestamp(sealed_block.timestamp), - )?; - - prague::ensure_well_formed_fields( - sealed_block.body(), - sidecar.prague(), - chain_spec.is_prague_active_at_timestamp(sealed_block.timestamp), - )?; + ensure_well_formed_fork_fields(&chain_spec, &sealed_block, &sidecar)?; Ok(sealed_block) } @@ -168,8 +153,7 @@ where ) .entered(); let transaction_count = raw_body.transactions.len(); - let mut slots: Vec> = Vec::new(); - slots.resize_with(transaction_count, || None); + let mut slots: Vec> = vec![None; transaction_count]; let mut received = 0usize; loop { @@ -212,13 +196,28 @@ where // The header hash was already computed and verified above, no need to reseal. let sealed_block = SealedBlock::new_unchecked(block, hash); + ensure_well_formed_fork_fields(&chain_spec, &sealed_block, &sidecar)?; + + Ok((sealed_block, transactions_root)) +} + +/// Validates the fork-specific fields of the block and sidecar (shanghai, cancun, prague). +fn ensure_well_formed_fork_fields( + chain_spec: &ChainSpec, + sealed_block: &SealedBlock>, + sidecar: &ExecutionPayloadSidecar, +) -> Result<(), PayloadError> +where + ChainSpec: EthereumHardforks, + T: SignedTransaction, +{ shanghai::ensure_well_formed_fields( sealed_block.body(), chain_spec.is_shanghai_active_at_timestamp(sealed_block.timestamp), )?; cancun::ensure_well_formed_fields( - &sealed_block, + sealed_block, sidecar.cancun(), chain_spec.is_cancun_active_at_timestamp(sealed_block.timestamp), )?; @@ -227,9 +226,7 @@ where sealed_block.body(), sidecar.prague(), chain_spec.is_prague_active_at_timestamp(sealed_block.timestamp), - )?; - - Ok((sealed_block, transactions_root)) + ) } /// Decodes raw payload transactions, mirroring the errors produced by From e621dfffacd2a0317876fcb6deab29e67757cdb2 Mon Sep 17 00:00:00 2001 From: Hesham Date: Fri, 14 Aug 2026 10:05:08 +0200 Subject: [PATCH 3/3] refactor(engine): drop raw tx-root passthrough from payload tx stream The header-derived transactions root always matches the raw bytes it was computed from, so reusing it in the pre-execution check would compare a value to itself and stop rejecting non-canonical transaction RLP. Return a plain SealedBlock, keep the re-encoding check, and lock the invariant in with tests. Also dedupes the body_tx send sites into a helper. Claude-Session: https://claude.ai/code/session_01QBN5jDc2WdmkaqM1r2GqTL --- crates/engine/primitives/src/lib.rs | 18 ++--- .../tree/src/tree/payload_processor/mod.rs | 29 +++++--- .../engine/tree/src/tree/payload_validator.rs | 25 ++++--- crates/engine/tree/src/tree/tests.rs | 67 ++++++++++++++----- crates/ethereum/node/src/engine.rs | 8 +-- crates/ethereum/payload/src/validator.rs | 60 +++++++++++------ 6 files changed, 133 insertions(+), 74 deletions(-) diff --git a/crates/engine/primitives/src/lib.rs b/crates/engine/primitives/src/lib.rs index 5402b9f24a1..70f55ffce09 100644 --- a/crates/engine/primitives/src/lib.rs +++ b/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}; @@ -237,21 +237,18 @@ pub trait PayloadValidator: Send + Sync + Unpin + 'static { /// payload and reproduce the exact errors of /// [`convert_payload_to_block`](Self::convert_payload_to_block). /// - /// Returns the block together with the transactions root computed from the raw payload - /// transaction bytes; `None` means the caller recomputes it from the block body. - /// /// The default drops the stream and delegates to - /// [`convert_payload_to_block`](Self::convert_payload_to_block), preserving today's - /// behavior for implementations that don't opt in via - /// [`supports_payload_tx_stream`](Self::supports_payload_tx_stream). + /// [`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, - ) -> Result<(SealedBlock, Option), NewPayloadError> { + ) -> Result, NewPayloadError> { drop(txs); - self.convert_payload_to_block(payload).map(|block| (block, None)) + self.convert_payload_to_block(payload) } /// Verifies payload post-execution w.r.t. hashed state updates. @@ -309,8 +306,7 @@ pub trait PayloadValidator: Send + Sync + Unpin + 'static { /// /// The index is required because the parallel decode fan-out may send transactions out of order. #[cfg(feature = "std")] -pub type PayloadTxStreamItem = - (usize, <::Body as reth_primitives_traits::BlockBody>::Transaction); +pub type PayloadTxStreamItem = (usize, <::Body as BlockBody>::Transaction); /// Receiving half of the decoded transaction stream consumed by /// [`PayloadValidator::convert_payload_to_block_with_tx_stream`]. diff --git a/crates/engine/tree/src/tree/payload_processor/mod.rs b/crates/engine/tree/src/tree/payload_processor/mod.rs index 353f5fb3ab7..4e2ed8579f1 100644 --- a/crates/engine/tree/src/tree/payload_processor/mod.rs +++ b/crates/engine/tree/src/tree/payload_processor/mod.rs @@ -286,9 +286,7 @@ where .for_each(|(idx, tx)| { let tx = tx.map(|tx| { let tx = WithTxEnv::new(tx); - if let Some(body_tx) = &body_tx { - let _ = body_tx.send((idx, tx.tx().clone())); - } + send_body_tx(body_tx.as_ref(), idx, &tx); let _ = prewarm_tx.send((idx, tx.clone())); tx }); @@ -342,9 +340,7 @@ where for (idx, tx) in chunk { if let Ok(tx) = &tx { - if let Some(body_tx) = &body_tx { - let _ = body_tx.send((idx, tx.tx().clone())); - } + send_body_tx(body_tx.as_ref(), idx, tx); let _ = prewarm_tx.send((idx, tx.clone())); } let _ = execute_tx.send((idx, tx)); @@ -528,9 +524,7 @@ fn convert_serial( let tx = convert.convert(raw_tx); let tx = tx.map(|tx| WithTxEnv::new(tx)); if let Ok(tx) = &tx { - if let Some(body_tx) = body_tx { - let _ = body_tx.send((idx, tx.tx().clone())); - } + send_body_tx(body_tx, idx, tx); let _ = prewarm_tx.send((idx, tx.clone())); } let _ = execute_tx.send((idx, tx)); @@ -538,6 +532,23 @@ fn convert_serial( } } +/// 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( + body_tx: Option<&mpsc::SyncSender<(usize, InnerTx)>>, + idx: usize, + tx: &WithTxEnv, +) where + InnerTx: Clone, + Recovered: reth_evm::RecoveredTx, +{ + 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>` with the diff --git a/crates/engine/tree/src/tree/payload_validator.rs b/crates/engine/tree/src/tree/payload_validator.rs index 6c2105b6d51..a7b14de29f8 100644 --- a/crates/engine/tree/src/tree/payload_validator.rs +++ b/crates/engine/tree/src/tree/payload_validator.rs @@ -940,11 +940,11 @@ where "convert_and_validate", ) .entered(); - let (block, tx_root) = match input { - BlockOrPayload::Block(block) => (block, None), + let block = match input { + BlockOrPayload::Block(block) => block, BlockOrPayload::Payload(payload) => match body_rx { Some(rx) => validator.convert_payload_to_block_with_tx_stream(payload, rx)?, - None => (validator.convert_payload_to_block(payload)?, None), + None => validator.convert_payload_to_block(payload)?, }, }; @@ -962,9 +962,7 @@ where } drop(_enter); - if let Err(e) = - consensus.validate_block_pre_execution_with_tx_root(&block, tx_root) - { + if let Err(e) = consensus.validate_block_pre_execution_with_tx_root(&block, None) { error!(target: "engine::tree::payload_validator", ?block, "Failed to validate block {}: {e}", block.hash()); return Err(InsertBlockError::consensus_error(e, block).into()) } @@ -2108,13 +2106,24 @@ impl PendingValidatedBlock { /// Blocks until conversion completes and returns a reference to the result. fn get(&mut self) -> &Result, InsertPayloadError> { - drop(self.body_tx.take()); + self.drop_sender(); self.handle.get() } /// Consumes the handle and returns the conversion result. fn into_inner(mut self) -> Result, InsertPayloadError> { - drop(self.body_tx.take()); + self.drop_sender(); self.handle.try_into_inner().expect("sole handle") } + + /// Disconnects the transaction stream so conversion can never block on transactions that + /// will not arrive. + fn drop_sender(&mut self) { + if self.body_tx.take().is_some() { + debug!( + target: "engine::tree::payload_validator", + "dropping payload tx sender before blocking on conversion" + ); + } + } } diff --git a/crates/engine/tree/src/tree/tests.rs b/crates/engine/tree/src/tree/tests.rs index 3bb7f093e8f..cbfa7a7f5f3 100644 --- a/crates/engine/tree/src/tree/tests.rs +++ b/crates/engine/tree/src/tree/tests.rs @@ -23,7 +23,9 @@ use alloy_rpc_types_engine::{ use assert_matches::assert_matches; use reth_chain_state::{test_utils::TestBlockBuilder, BlockState}; use reth_chainspec::{ChainSpec, HOLESKY, MAINNET}; -use reth_engine_primitives::{EngineApiValidator, ForkchoiceStatus, NoopInvalidBlockHook}; +use reth_engine_primitives::{ + EngineApiValidator, ForkchoiceStatus, NoopInvalidBlockHook, PayloadValidator, +}; use reth_ethereum_consensus::EthBeaconConsensus; use reth_ethereum_engine_primitives::{EthEngineTypes, EthPayloadAttributes}; use reth_ethereum_primitives::{Block, EthPrimitives, TransactionSigned}; @@ -3302,11 +3304,20 @@ fn test_payload_tx_stream_malformed_tx_reports_decode_error() { streaming_payload(&genesis, vec![Bytes::from_static(b"garbage")], genesis.gas_limit); let err = harness.validate_payload(payload).unwrap_err(); - match err { - InsertPayloadError::Payload(reth_payload_primitives::NewPayloadError::Eth( - PayloadError::Decode(_), - )) => {} - other => panic!("expected decode error, got: {other:?}"), + match &err { + InsertPayloadError::Payload(payload_err) => { + assert!( + matches!( + payload_err, + reth_payload_primitives::NewPayloadError::Eth(PayloadError::Decode(_)) + ), + "expected decode error, got: {payload_err:?}" + ); + // A decode error is not a hash mismatch, so the engine reports the parent as + // `latestValidHash` — unchanged from the non-streaming path. + assert!(!payload_err.is_block_hash_mismatch()); + } + other => panic!("expected payload error, got: {other:?}"), } } @@ -3326,11 +3337,21 @@ fn test_payload_tx_stream_block_hash_takes_precedence() { let payload = ExecutionData { payload: payload.into(), sidecar }; let err = harness.validate_payload(payload).unwrap_err(); - match err { - InsertPayloadError::Payload(reth_payload_primitives::NewPayloadError::Eth( - PayloadError::BlockHash { .. }, - )) => {} - other => panic!("expected block hash mismatch, got: {other:?}"), + match &err { + InsertPayloadError::Payload(payload_err) => { + assert!( + matches!( + payload_err, + reth_payload_primitives::NewPayloadError::Eth(PayloadError::BlockHash { .. }) + ), + "expected block hash mismatch, got: {payload_err:?}" + ); + // This is the point of the check-ordering flip: reporting the hash mismatch is what + // makes the engine answer `latestValidHash: null` per the engine API, where the + // non-streaming path's decode error would have reported the parent instead. + assert!(payload_err.is_block_hash_mismatch()); + } + other => panic!("expected payload error, got: {other:?}"), } } @@ -3369,12 +3390,22 @@ fn test_payload_tx_stream_aborted_execution_does_not_hang() { let payload = streaming_payload(&genesis, vec![streaming_raw_tx(0)], genesis.gas_limit); - // The crafted payload cannot pass consensus/execution. A block-level error proves the - // streamed transactions were assembled into a block (conversion terminated) rather than - // validation waiting on the tx stream forever or failing to decode. + // The streaming path must assemble the same block the non-streaming path would, whether the + // transactions arrived over the stream or the disconnect fallback re-decoded them. Comparing + // against the non-streaming conversion locks that equivalence; the error kind alone would + // pass even if the block came out wrong. + let expected = PayloadValidator::::convert_payload_to_block( + &EthereumEngineValidator::new(MAINNET.clone()), + payload.clone(), + ) + .expect("payload converts via the non-streaming path"); + + // The crafted payload cannot pass consensus/execution. A block-level error carries the + // assembled block, proving conversion terminated rather than waiting on the tx stream + // forever or failing to decode. let err = harness.validate_payload(payload).unwrap_err(); - assert!( - matches!(err, InsertPayloadError::Block(_)), - "expected post-conversion error, got: {err:?}" - ); + match err { + InsertPayloadError::Block(err) => assert_eq!(err.block(), &expected), + other => panic!("expected post-conversion block error, got: {other:?}"), + } } diff --git a/crates/ethereum/node/src/engine.rs b/crates/ethereum/node/src/engine.rs index 43c1df0fc25..d20654d96c1 100644 --- a/crates/ethereum/node/src/engine.rs +++ b/crates/ethereum/node/src/engine.rs @@ -1,6 +1,5 @@ //! Validates execution payload wrt Ethereum Execution Engine API version. -use alloy_primitives::B256; use alloy_rpc_types_engine::ExecutionData; pub use alloy_rpc_types_engine::{ ExecutionPayloadEnvelopeV2, ExecutionPayloadEnvelopeV3, ExecutionPayloadEnvelopeV4, @@ -59,11 +58,8 @@ where &self, payload: ExecutionData, txs: PayloadTxStream, - ) -> Result<(SealedBlock, Option), NewPayloadError> { - self.inner - .ensure_well_formed_payload_with_tx_stream(payload, txs) - .map(|(block, tx_root)| (block, Some(tx_root))) - .map_err(Into::into) + ) -> Result, NewPayloadError> { + self.inner.ensure_well_formed_payload_with_tx_stream(payload, txs).map_err(Into::into) } } diff --git a/crates/ethereum/payload/src/validator.rs b/crates/ethereum/payload/src/validator.rs index fd50d5d19fb..b54590239e7 100644 --- a/crates/ethereum/payload/src/validator.rs +++ b/crates/ethereum/payload/src/validator.rs @@ -1,7 +1,7 @@ //! Validates execution payload wrt Ethereum consensus rules use alloy_consensus::Block; -use alloy_primitives::{Bytes, B256}; +use alloy_primitives::Bytes; use alloy_rpc_types_engine::{ExecutionData, ExecutionPayloadSidecar, PayloadError}; use reth_chainspec::EthereumHardforks; use reth_payload_validator::{cancun, prague, shanghai}; @@ -49,7 +49,7 @@ impl EthereumExecutionPayloadValidator &self, payload: ExecutionData, txs: mpsc::Receiver<(usize, T)>, - ) -> Result<(SealedBlock>, B256), PayloadError> { + ) -> Result>, PayloadError> { ensure_well_formed_payload_with_tx_stream(&self.chain_spec, payload, txs) } } @@ -116,14 +116,11 @@ where /// If `txs` disconnects before all transactions arrive (e.g. a malformed transaction stopped the /// decoder, or execution was aborted), this falls back to decoding the retained raw transaction /// bytes, reproducing the errors of [`ensure_well_formed_payload`]. -/// -/// Returns the sealed block together with the transactions root computed from the raw payload -/// bytes, so callers can skip re-encoding transactions for root validation. pub fn ensure_well_formed_payload_with_tx_stream( chain_spec: ChainSpec, payload: ExecutionData, txs: mpsc::Receiver<(usize, T)>, -) -> Result<(SealedBlock>, B256), PayloadError> +) -> Result>, PayloadError> where ChainSpec: EthereumHardforks, T: SignedTransaction, @@ -132,10 +129,9 @@ where let expected_hash = payload.block_hash(); - // Build the block with raw transaction bytes; the transactions root in the header is - // computed directly from the raw bytes, so transactions are never re-encoded. + // Build the block with raw transaction bytes so the body can be filled from the stream. The + // raw bytes are retained for the disconnect fallback. let raw_block = payload.into_block_with_sidecar_raw(&sidecar)?; - let transactions_root = raw_block.header.transactions_root; // Ensure the hash included in the payload matches the block hash before waiting on any // transactions, so header validation overlaps with the execution-side decode. @@ -147,11 +143,7 @@ where let Block { header, body: raw_body } = raw_block; let transactions = { - let _span = debug_span!( - target: "engine::tree::payload_validator", - "assemble_body_from_stream", - ) - .entered(); + let _span = debug_span!(target: "payload_builder", "assemble_body_from_stream").entered(); let transaction_count = raw_body.transactions.len(); let mut slots: Vec> = vec![None; transaction_count]; let mut received = 0usize; @@ -174,7 +166,7 @@ where // aborted execution, or an engine early-return). Decode from the retained // raw bytes to reproduce the non-streaming behavior. debug!( - target: "engine::tree::payload_validator", + target: "payload_builder", received, transaction_count, "payload tx stream disconnected, falling back to full decode" @@ -198,7 +190,7 @@ where ensure_well_formed_fork_fields(&chain_spec, &sealed_block, &sidecar)?; - Ok((sealed_block, transactions_root)) + Ok(sealed_block) } /// Validates the fork-specific fields of the block and sidecar (shanghai, cancun, prague). @@ -244,7 +236,10 @@ fn decode_transactions(raw: &[Bytes]) -> Result, Pa #[cfg(test)] mod tests { use super::*; - use alloy_consensus::{BlockBody, SignableTransaction, TxLegacy}; + use alloy_consensus::{ + proofs::ordered_trie_root_encoded, BlockBody, SignableTransaction, TxLegacy, + }; + use alloy_eips::eip2718::Encodable2718; use alloy_primitives::{Address, Signature, TxKind, B256, U256}; use alloy_rpc_types_engine::{ExecutionPayloadSidecar, ExecutionPayloadV1}; use reth_chainspec::MAINNET; @@ -275,7 +270,6 @@ mod tests { } fn payload_with_txs(txs: &[TransactionSigned]) -> ExecutionData { - use alloy_eips::eip2718::Encodable2718; payload_with_raw_txs(txs.iter().map(|tx| tx.encoded_2718().into()).collect()) } @@ -293,10 +287,8 @@ mod tests { } drop(tx); - let (block, tx_root) = - ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap(); + let block = ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap(); assert_eq!(block, expected); - assert_eq!(tx_root, expected.transactions_root); } #[test] @@ -312,7 +304,7 @@ mod tests { tx.send((0, txs[0].clone())).unwrap(); drop(tx); - let (block, _) = ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap(); + let block = ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap(); assert_eq!(block, expected); } @@ -334,6 +326,30 @@ mod tests { assert!(matches!(err, PayloadError::Decode(_))); } + /// The header's transactions root is derived from the *raw* payload bytes, so it must never + /// be reused as the pre-execution check's calculated root: that check compares the raw bytes + /// against the re-encoded decoded body, which is what rejects non-canonical transaction RLP. + /// Passing the header value through would make the comparison compare a value to itself. + #[test] + fn header_tx_root_is_not_a_substitute_for_re_encoding() { + let tx = signed_tx(0); + let canonical = tx.encoded_2718(); + + // Re-encode the transaction's RLP payload length with a longer-than-minimal header. The + // bytes decode to the same transaction but are not what re-encoding it produces. + let mut non_canonical = canonical.clone(); + non_canonical.extend_from_slice(&[0x00]); + + let payload = payload_with_raw_txs(vec![non_canonical.clone().into()]); + let header_root = + payload.payload.into_v1().into_block_raw().unwrap().header.transactions_root; + + // The header root always matches the raw bytes it was derived from, whatever they are, + // so on its own it proves nothing about the decoded body. + assert_eq!(header_root, ordered_trie_root_encoded(&[Bytes::from(non_canonical)])); + assert_ne!(header_root, ordered_trie_root_encoded(&[Bytes::from(canonical)])); + } + #[test] fn block_hash_checked_before_transactions() { // Both a bad hash and a malformed transaction: the streaming path reports the hash