diff --git a/crates/engine/primitives/src/lib.rs b/crates/engine/primitives/src/lib.rs index 7855b803871..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}; @@ -212,6 +212,45 @@ 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). + /// + /// 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, + ) -> Result, 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: 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 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..4e2ed8579f1 100644 --- a/crates/engine/tree/src/tree/payload_processor/mod.rs +++ 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 = 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 = PayloadTxStreamSender::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>( &self, @@ -173,6 +179,7 @@ where hint_stream: Option, hashed_update_stream: Option, parallel_bal_execution: bool, + body_tx: Option>, ) -> IteratorPayloadHandle 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>, ) -> (IteratorPrewarmTxReceiver, IteratorExecuteTxReceiver) { 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,15 +512,19 @@ 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 { + send_body_tx(body_tx, idx, tx); let _ = prewarm_tx.send((idx, tx.clone())); } let _ = execute_tx.send((idx, tx)); @@ -502,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 9c4dde82d67..a7b14de29f8 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::{ @@ -142,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, @@ -496,10 +497,22 @@ 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 (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()); + 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 @@ -508,7 +521,7 @@ where match $expr { Ok(val) => val, Err(e) => { - let block = validated_block.try_into_inner().expect("sole handle")?; + let block = validated_block.into_inner()?; return Err(InsertBlockError::new(block, e.into()).into()) } } @@ -536,10 +549,7 @@ where { // 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")) } } @@ -551,7 +561,7 @@ where else { // this is pre-validated in the tree return Err(InsertBlockError::new( - validated_block.try_into_inner().expect("sole handle")?, + validated_block.into_inner()?, ProviderError::HeaderNotFound(parent_hash.into()).into(), ) .into()) @@ -635,6 +645,7 @@ where hint_stream, hashed_update_stream, parallel_bal_execution, + validated_block.take_sender(), )); // Create optional cache stats for detailed block logging @@ -763,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. @@ -903,11 +914,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>, @@ -926,9 +942,10 @@ where .entered(); let block = match input { BlockOrPayload::Block(block) => block, - BlockOrPayload::Payload(payload) => { - validator.convert_payload_to_block(payload)? - } + 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)?, + }, }; if let Err(e) = consensus.validate_header(block.sealed_header()) { @@ -945,9 +962,7 @@ where } drop(_enter); - if let Err(e) = - consensus.validate_block_pre_execution_with_tx_root(&block, None) - { + 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()) } @@ -1350,6 +1365,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 +1384,7 @@ where hint_stream: Option, hashed_update_stream: Option, parallel_bal_execution: bool, + body_tx: Option>, ) -> Result< PayloadHandle< impl ExecutableTxFor + use, @@ -1384,6 +1401,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()); @@ -2069,3 +2087,43 @@ 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> { + self.drop_sender(); + self.handle.get() + } + + /// Consumes the handle and returns the conversion result. + fn into_inner(mut self) -> Result, InsertPayloadError> { + 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 53cbf1d4e5f..cbfa7a7f5f3 100644 --- a/crates/engine/tree/src/tree/tests.rs +++ b/crates/engine/tree/src/tree/tests.rs @@ -9,24 +9,28 @@ 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}; 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}; -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}; @@ -3195,3 +3199,213 @@ 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 [`EthereumEngineValidator`], which opts into the +/// payload tx stream. +struct StreamingValidatorHarness { + harness: TestHarness, + validator: BasicEngineValidator, +} + +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, + EthEvmConfig::new(chain_spec.clone()), + 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: 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: 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 { + 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(), + }; + let signed: TransactionSigned = tx.into_signed(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(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:?}"), + } +} + +/// 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(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:?}"), + } +} + +/// 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 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(); + 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 f1b880ab253..d20654d96c1 100644 --- a/crates/ethereum/node/src/engine.rs +++ b/crates/ethereum/node/src/engine.rs @@ -6,7 +6,7 @@ pub use alloy_rpc_types_engine::{ 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 +49,18 @@ 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, NewPayloadError> { + self.inner.ensure_well_formed_payload_with_tx_stream(payload, txs).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..b54590239e7 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_rpc_types_engine::{ExecutionData, PayloadError}; +use alloy_primitives::Bytes; +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}; -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>, 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 @@ -86,13 +100,116 @@ where }) } + ensure_well_formed_fork_fields(&chain_spec, &sealed_block, &sidecar)?; + + 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`]. +pub fn ensure_well_formed_payload_with_tx_stream( + chain_spec: ChainSpec, + payload: ExecutionData, + txs: mpsc::Receiver<(usize, T)>, +) -> Result>, PayloadError> +where + ChainSpec: EthereumHardforks, + T: SignedTransaction, +{ + let ExecutionData { payload, sidecar } = payload; + + let expected_hash = payload.block_hash(); + + // 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)?; + + // 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: "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; + + 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: "payload_builder", + 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); + + ensure_well_formed_fork_fields(&chain_spec, &sealed_block, &sidecar)?; + + Ok(sealed_block) +} + +/// 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), )?; @@ -101,7 +218,152 @@ where sealed_block.body(), sidecar.prague(), chain_spec.is_prague_active_at_timestamp(sealed_block.timestamp), - )?; + ) +} - Ok(sealed_block) +/// 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::{ + 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; + 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 { + 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 = ensure_well_formed_payload_with_tx_stream(&*MAINNET, payload, rx).unwrap(); + assert_eq!(block, expected); + } + + #[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(_))); + } + + /// 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 + // 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 { .. })); + } }