From 4b7797bcf792b68ace6143e66ca23dda56512446 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Fri, 2 Oct 2026 17:09:54 -0400 Subject: [PATCH] Emit Firehose init and genesis without an ExEx The Firehose ExEx only emitted FIRE INIT and the genesis block; its block tracing was disabled. Emit both from the component-initialized hook, which runs before the consensus engine exists, so no block is traced first. --- CHANGELOG.md | 15 ++ Cargo.lock | 3 - bin/reth/src/main.rs | 4 +- crates/firehose/Cargo.toml | 3 - crates/firehose/src/block_tracer.rs | 2 +- crates/firehose/src/init.rs | 77 +++++++ crates/firehose/src/lib.rs | 8 +- crates/firehose/src/mapper.rs | 8 +- crates/firehose/src/runner.rs | 330 ---------------------------- 9 files changed, 99 insertions(+), 351 deletions(-) create mode 100644 crates/firehose/src/init.rs delete mode 100644 crates/firehose/src/runner.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c0c166a21e..c37f1cd12b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ This changelog covers Firehose-specific changes only. For upstream reth changes, The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). +## Unreleased + +### Changed + +- `reth` no longer installs a Firehose execution extension (ExEx). It only emitted `FIRE INIT` + and the genesis block; blocks are traced by the engine. Both are now emitted at startup before + the consensus engine exists, so no block can be traced ahead of `FIRE INIT`, and the node + stops writing the ExEx write-ahead log. + +### Removed + +- `reth_firehose::run_exex` and `reth_firehose::emit_genesis_block_if_empty`. Call + `reth_firehose::init_blockchain(&provider)` from the node builder's `on_component_initialized` + hook instead. `reth-firehose` no longer depends on `reth-exex`. + ## reth-v2.7.0-fh3.1 ### Changed diff --git a/Cargo.lock b/Cargo.lock index a79b6f71a0d..897ac91d851 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9254,16 +9254,13 @@ dependencies = [ "base64 0.22.1", "eyre", "firehose-tracer", - "futures", "prost", "rayon", "reth-chainspec", - "reth-ethereum-forks", "reth-ethereum-primitives", "reth-evm", "reth-evm-ethereum", "reth-execution-types", - "reth-exex", "reth-node-api", "reth-primitives-traits", "reth-provider", diff --git a/bin/reth/src/main.rs b/bin/reth/src/main.rs index 5080b023093..ad875c75dd8 100644 --- a/bin/reth/src/main.rs +++ b/bin/reth/src/main.rs @@ -50,9 +50,7 @@ fn main() { EthereumNode::components().executor(FirehoseExecutorBuilder::default()), ) .with_add_ons(EthereumAddOns::default()) - .install_exex("firehose", |ctx| async move { - Ok(async move { reth_firehose::run_exex(ctx).await }) - }) + .on_component_initialized(|node| reth_firehose::init_blockchain(&node.provider)) .launch_with_debug_capabilities() .await?; diff --git a/crates/firehose/Cargo.toml b/crates/firehose/Cargo.toml index 351f8c9ee61..2cba3ee7410 100644 --- a/crates/firehose/Cargo.toml +++ b/crates/firehose/Cargo.toml @@ -15,11 +15,9 @@ firehose-tracer.workspace = true # reth reth-chainspec.workspace = true -reth-ethereum-forks.workspace = true reth-evm.workspace = true reth-evm-ethereum.workspace = true reth-execution-types.workspace = true -reth-exex.workspace = true reth-node-api.workspace = true reth-ethereum-primitives.workspace = true reth-primitives-traits.workspace = true @@ -38,7 +36,6 @@ alloy-rlp.workspace = true # std eyre.workspace = true -futures.workspace = true rayon = { workspace = true, optional = true } [dev-dependencies] diff --git a/crates/firehose/src/block_tracer.rs b/crates/firehose/src/block_tracer.rs index b91114e4f5e..e92c9b037a6 100644 --- a/crates/firehose/src/block_tracer.rs +++ b/crates/firehose/src/block_tracer.rs @@ -54,7 +54,7 @@ impl FirehoseBlockTracer { /// /// The genesis block (block 0) never goes through this guard: reth writes it directly to /// the database without executing it, and it is emitted standalone via - /// [`crate::runner::emit_genesis_block_if_empty`] at startup. + /// [`crate::init::emit_genesis_block_on_empty_chain`] at startup. /// /// Takes a [`SealedBlock`] rather than a `RecoveredBlock` so the guard can be started before /// transaction senders have been recovered. Block-level data read by the mapper is signer-free. diff --git a/crates/firehose/src/init.rs b/crates/firehose/src/init.rs new file mode 100644 index 00000000000..588e5aa4e46 --- /dev/null +++ b/crates/firehose/src/init.rs @@ -0,0 +1,77 @@ +use crate::{mapper, prelude::*}; +use reth_chainspec::EthChainSpec; +use reth_provider::{BlockReader, ChainSpecProvider}; + +/// Emits `FIRE INIT` and, on an empty chain, the genesis block. +/// +/// Call it once from the node builder's `on_component_initialized` hook. The tracer panics on a +/// block that starts before `FIRE INIT`, and the consensus engine can execute blocks before the +/// `on_node_started` hook runs. +pub fn init_blockchain

(provider: &P) -> eyre::Result<()> +where + P: BlockReader + ChainSpecProvider, + ::Header: BlockHeader + alloy_primitives::Sealable, + <::Body as reth_primitives_traits::BlockBody>::OmmerHeader: + BlockHeader + alloy_primitives::Sealable, +{ + let chain_spec = provider.chain_spec(); + crate::tracer().on_blockchain_init( + "reth", + env!("CARGO_PKG_VERSION"), + firehose_tracer::config::ChainConfig::new(chain_spec.chain().id()), + ); + + emit_genesis_block_on_empty_chain(provider, chain_spec.genesis()) +} + +/// Emits the genesis block (block 0) to the Firehose stream if the chain is empty (head still +/// at genesis). +/// +/// Reth never executes the genesis block — it is written straight to the database during init — +/// so no execution hook ever fires for it. Without this, a Firehose stream started from scratch +/// begins at block 1 and downstream consumers waiting for block 0 stall forever. Mirrors geth's +/// `OnGenesisBlock` hook: emitted with the full genesis alloc, only when starting on an empty +/// chain. +/// +/// Call it once at node startup, after `on_blockchain_init` and before any block is traced. +/// Embedders that report their own node name or version in `FIRE INIT` call this directly instead +/// of [`init_blockchain`]. +pub fn emit_genesis_block_on_empty_chain

( + provider: &P, + genesis: &alloy_genesis::Genesis, +) -> eyre::Result<()> +where + P: BlockReader, + ::Header: BlockHeader + alloy_primitives::Sealable, + <::Body as reth_primitives_traits::BlockBody>::OmmerHeader: + BlockHeader + alloy_primitives::Sealable, +{ + use reth_primitives_traits::Block as _; + + // OP-stack chains may place their genesis at a non-zero height (e.g. OP mainnet's + // post-bedrock genesis); `genesis.number` carries it, defaulting to 0. + let genesis_number = genesis.number.unwrap_or(0); + if provider.best_block_number()? != genesis_number { + return Ok(()); + } + + let genesis_block = provider + .block(genesis_number.into())? + .ok_or_else(|| eyre::eyre!("genesis block {genesis_number} not found in database"))? + .seal_slow(); + + crate::tracer().on_genesis_block( + firehose_tracer::types::BlockEvent { + block: mapper::to_block_data(&genesis_block), + // Genesis is finalized by definition. + finalized: Some(firehose_tracer::types::FinalizedBlockRef { + number: genesis_number, + hash: Some(genesis_block.hash()), + }), + flash_block: None, + }, + mapper::to_genesis_alloc(genesis), + ); + + Ok(()) +} diff --git a/crates/firehose/src/lib.rs b/crates/firehose/src/lib.rs index aceea7d5ffb..e92e900da28 100644 --- a/crates/firehose/src/lib.rs +++ b/crates/firehose/src/lib.rs @@ -1,6 +1,6 @@ //! Firehose crate providing blockchain data processing modules. //! -//! This crate contains modules for inspection, mapping, prelude utilities, and running tasks. +//! This crate contains modules for inspection, mapping, prelude utilities, and node startup. /// Block-level drop guard that manages the Firehose tracer lifecycle across validation. pub mod block_tracer; @@ -8,14 +8,14 @@ pub mod block_tracer; pub mod executor; /// Resolves which finalized block a Firehose block event may advertise. pub mod finality; +/// Startup module emitting `FIRE INIT` and the genesis block. +pub mod init; /// Inspector module for analyzing blockchain data. pub mod inspector; /// Mapper module for transforming blockchain data. pub mod mapper; /// Prelude module with common imports and utilities. pub mod prelude; -/// Runner module for executing processing tasks. -pub mod runner; pub use block_tracer::{FirehoseBlockTracer, GlobalTracerGuard}; pub use executor::{ @@ -25,8 +25,8 @@ pub use executor::{ }; pub use finality::finalized_ref_for_block; pub use firehose_tracer::types::FinalizedBlockRef; +pub use init::{emit_genesis_block_on_empty_chain, init_blockchain}; pub use inspector::{FramelessTx, PostTxGasAccounting}; -pub use runner::{emit_genesis_block_if_empty, emit_genesis_block_on_empty_chain, run_exex}; use std::{ io::Write, diff --git a/crates/firehose/src/mapper.rs b/crates/firehose/src/mapper.rs index 949ec3620e5..fde98df0ce5 100644 --- a/crates/firehose/src/mapper.rs +++ b/crates/firehose/src/mapper.rs @@ -15,7 +15,7 @@ use firehose_tracer::types::{ AccessTuple, BlockData, GenesisAlloc, SetCodeAuthorization, TxEvent, UncleData, WithdrawalData, }; use reth_primitives_traits::{Block as BlockTrait, BlockBody}; -use reth_provider::{AccountReader, ProviderResult}; +use reth_provider::AccountReader; pub(crate) fn to_genesis_alloc(genesis: &Genesis) -> GenesisAlloc { genesis @@ -124,12 +124,6 @@ where .collect() } -pub(crate) fn to_finalized_ref( - block_ref: ProviderResult>, -) -> Option { - finalized_ref_from_num_hash(block_ref.ok().flatten()) -} - /// Builds a Firehose [`FinalizedBlockRef`](firehose_tracer::types::FinalizedBlockRef) from the /// finalized block's number/hash, if known. /// diff --git a/crates/firehose/src/runner.rs b/crates/firehose/src/runner.rs deleted file mode 100644 index 8094cb8cb52..00000000000 --- a/crates/firehose/src/runner.rs +++ /dev/null @@ -1,330 +0,0 @@ -use crate::{inspector, mapper, prelude::*}; -use alloy_consensus::transaction::TxHashRef as _; -use alloy_primitives::Bytes; -use eyre::Context; -use futures::StreamExt; -use reth_chainspec::EthChainSpec; -use reth_ethereum_forks::EthereumHardforks; -use reth_evm::execute::BlockExecutor; -use reth_exex::{ExExContext, ExExEvent}; -use reth_provider::{BlockIdReader, BlockReader, EvmStateProviderBox, StateProviderFactory}; -use reth_revm::{ - database::StateProviderDatabase, - revm::{context::Block as _, Database as _}, - State, -}; - -/// Executes EVM transactions in a block one by one, firing tracer hooks at the appropriate times. -/// -/// This is a re-implementation for usage within ExEx and compatible with Firehose Geth Live -/// Tracing. -pub fn trace_block( - ctx: &ExExContext, - evm_config: &Node::Evm, - block: &RecoveredBlock, - receipts: &Vec>, - get_signature: &F, - shared_state: &mut State>, -) -> eyre::Result<()> -where - ChainSpec: EthereumHardforks + EthChainSpec, - F: Fn(&SignedTx) -> (B256, B256, Bytes), -{ - use alloy_consensus::TxReceipt; - - let tracer = &mut *crate::tracer(); - - tracer.on_block_start(firehose_tracer::types::BlockEvent { - block: mapper::to_block_data(block.sealed_block()), - finalized: mapper::to_finalized_ref(ctx.provider().finalized_block_num_hash()), - flash_block: None, - }); - - let evm_env = evm_config - .evm_env(block.header()) - .wrap_err_with(|| format!("Failed to build EVM env for block {}", block.number()))?; - // context_for_block type-checks because FullNodeComponents bounds Node::Evm: - // ConfigureEvm::Primitives> - let exec_ctx = evm_config - .context_for_block(block.sealed_block()) - .wrap_err_with(|| format!("Failed to build EVM context for block {}", block.number()))?; - - // Inspector borrows tracer mutably for the duration of the block execution. - // All tracer lifecycle calls (on_tx_start, on_tx_end, etc.) go through - // executor.evm_mut().inspector_mut().tracer_mut() while the inspector is live. - let inspector = inspector::FirehoseInspector::new(tracer); - // The EVM borrows `shared_state` mutably for the duration of this block. Bundle updates - // from prior blocks in the same ChainCommitted notification are already reflected in its - // cache via earlier `commit_transaction` calls, so nonce / balance reads here see the - // correct pre-block state without re-querying the historical provider. - let evm = evm_config.evm_with_env_and_inspector(&mut *shared_state, evm_env, inspector); - let mut executor = evm_config.create_executor(evm, exec_ctx); - - // AFAIK this is actually unused for tracing, commented out to try, will delete if conclusive - // // Set state hook to log EvmState changes after each transaction and system call - // executor.set_state_hook(Some(Box::new( - // |source: reth_evm::block::StateChangeSource, state: &reth::revm::revm::state::EvmState| { - // inspector::log_evm_state(&format!("state_hook({source:?})"), state); - // }, - // ))); - - // System calls (EIP-4788, EIP-2935, etc.) — handled by apply_pre_execution_changes - executor.evm_mut().inspector_mut().tracer_mut().on_system_call_start(); - executor.apply_pre_execution_changes().wrap_err_with(|| { - format!("Failed to apply pre-execution changes for block {}", block.number()) - })?; - executor.evm_mut().inspector_mut().tracer_mut().on_system_call_end(); - - let mut prev_cumulative_gas: u64 = 0; - let mut log_index: u32 = 0; - - for (tx_index, (recovered_tx, receipt)) in - block.transactions_recovered().zip(receipts.iter()).enumerate() - { - let tx: &SignedTx = &**recovered_tx; - let (r, s, v) = get_signature(tx); - let tx_event = mapper::signed_tx_to_tx_event(tx, recovered_tx.signer(), tx_index, r, s, v); - - // Fresh state reader per transaction for on_tx_start StateReader. - // - // KNOWN LIMITATION: this reader is resolved from the provider at `parent_hash`, not - // from the live `shared_state`. The firehose StateReader observations it produces are - // pre-block, not pre-tx. EVM-level nonce validation is unaffected (that uses the - // shared_state cache, not this reader). - //let state_reader_provider = ctx - // .provider() - // .state_by_block_hash(parent_hash) - // .wrap_err_with(|| { - // format!( - // "Failed to get state reader for block {} tx_index={tx_index} tx_hash={}", - // block.number(), - // recovered_tx.tx_hash() - // ) - // })?; - //let state_reader = Box::new(mapper::StateReaderAdapter(state_reader_provider)); - - executor.evm_mut().inspector_mut().tracer_mut().on_tx_start(tx_event, None); - - let caller_nonce = executor - .evm_mut() - .db_mut() - .basic(recovered_tx.signer())? - .ok_or_else(|| { - eyre::eyre!( - "Failed to get caller account info for block {} tx_index={tx_index} tx_hash={}", - block.number(), - recovered_tx.tx_hash() - ) - })? - .nonce; - info!(target: "firehose", block = block.number(), tx_index, tx_hash = ?recovered_tx.tx_hash(), caller_nonce, "Executing transaction"); - - // execute_transaction_without_commit runs the full EVM execution (including po - // gas refund and miner fee) and returns the final EvmState without committing to DB. - // Inspector hooks (on_call_enter, on_call_exit, on_opcode, etc.) fire during transact(). - let tx_result = - executor.execute_transaction_without_commit(recovered_tx).wrap_err_with(|| { - format!( - "Failed to execute transaction block={} tx_index={tx_index} tx_hash={}", - block.number(), - recovered_tx.tx_hash() - ) - })?; - - // Emit post-execution balance changes (gas refund to sender, miner fee to coinbase). - // revm's post_execution runs reimburse_caller and reward_beneficiary after the last - // inspector hook, so we explicitly compute and emit them using Ethereum gas rules. - { - let result_gas_used = { - use alloy_evm::block::TxResult as _; - tx_result.result().result.tx_gas_used() - }; - let sender = recovered_tx.signer(); - let coinbase = block.header().beneficiary(); - let gas_limit = tx.gas_limit(); - let base_fee = block.header().base_fee_per_gas().unwrap_or(0); - let effective_gas_price: u128 = if tx.is_dynamic_fee() { - std::cmp::min( - tx.max_fee_per_gas(), - base_fee as u128 + tx.max_priority_fee_per_gas().unwrap_or(0), - ) - } else { - tx.gas_price().unwrap_or(0) - }; - - // Committed log count from the ExecutionResult (empty on Revert/Halt). - let committed_log_count = { - use alloy_evm::block::TxResult as _; - tx_result.result().result.logs().len() as u32 - }; - let (db, inspector, _) = executor.evm_mut().components_mut(); - inspector.process_post_tx_balance_changes( - sender, - coinbase, - gas_limit, - result_gas_used, - effective_gas_price, - base_fee, - committed_log_count, - |addr| db.basic(addr).ok().flatten().map(|info| info.balance).unwrap_or(U256::ZERO), - ); - } - - executor.commit_transaction(tx_result); - - let cumulative_gas = receipt.cumulative_gas_used(); - let gas_used = cumulative_gas - prev_cumulative_gas; - let log_count = receipt.logs().len() as u32; - // EIP-4844 receipt fields. `blob_gas_used()` returns `None` for non-blob tx types; - // `blob_gasprice()` returns `None` for pre-Cancun blocks. Matches Geth's receipt shape. - let blob_gas_used = tx.blob_gas_used().unwrap_or(0); - let blob_gas_price = executor.evm().block().blob_gasprice().map(U256::from); - let receipt_data = mapper::to_receipt_data( - receipt, - tx_index as u32, - gas_used, - log_index, - blob_gas_used, - blob_gas_price, - ); - prev_cumulative_gas = cumulative_gas; - log_index += log_count; - - executor.evm_mut().inspector_mut().tracer_mut().on_tx_end(Some(&receipt_data), None); - } - - executor.evm_mut().inspector_mut().tracer_mut().on_system_call_start(); - - // Post-execution changes (block rewards, withdrawals, etc.) - // This consumes the executor, dropping the inspector and releasing the tracer borrow. - // State mutations (pre-execution changes, tx commits, post-execution changes) remain in - // `shared_state`, ready for the next block in this chain notification. - executor.apply_post_execution_changes().wrap_err_with(|| { - format!("Failed to apply post-execution changes for block {}", block.number()) - })?; - - // Tracer borrow released — can call directly again - tracer.on_system_call_end(); - - tracer.on_block_end(None); - - Ok(()) -} - -/// ExEx entry point for Firehose live-block tracing. -/// -/// Loops over `ChainCommitted` notifications, re-executes each block with `trace_block`, and -/// signals `FinishedHeight` after each block so the WAL can be pruned. -pub async fn run_exex(mut ctx: ExExContext) -> eyre::Result<()> -where - Node: FullNodeComponents, - Node::Provider: BlockReader + StateProviderFactory, - ChainSpec: EthereumHardforks + EthChainSpec, - SignedTx: mapper::SignatureFields, -{ - let chain_id = ctx.config.chain.chain().id(); - crate::tracer().on_blockchain_init( - "reth", - env!("CARGO_PKG_VERSION"), - firehose_tracer::config::ChainConfig::new(chain_id), - ); - - emit_genesis_block_if_empty(&ctx)?; - - while let Some(notification) = ctx.notifications.next().await { - let notification = notification?; - - if let Some(committed) = notification.committed_chain() { - info!(chain = ?committed.range(), "Chain committed, tracing {} blocks", committed.len()); - //let first_block = committed.first(); - //let parent_hash = first_block.parent_hash(); - - //let state_provider = - // ctx.provider().state_by_block_hash(parent_hash).wrap_err_with(|| { - // format!("Failed to get state provider for parent block {}", parent_hash) - // })?; - - //let mut _shared_state = State::builder() - // .with_database(StateProviderDatabase::new(state_provider)) - // .with_bundle_update() - // .build(); - - //let _evm_config = ctx.evm_config().clone(); - - for (block, _receipts) in committed.blocks_and_receipts() { - // trace_block( - // &ctx, - // &evm_config, - // block, - // receipts, - // &|tx: &SignedTx| tx.signature_fields(), - // &mut shared_state, - // )?; - - ctx.events.send(ExExEvent::FinishedHeight(block.num_hash()))?; - } - } - } - - Ok(()) -} - -/// Emits the genesis block (block 0) to the Firehose stream if the chain is empty (head still -/// at genesis). -/// -/// Reth never executes the genesis block — it is written straight to the database during init — -/// so no execution hook ever fires for it. Without this, a Firehose stream started from scratch -/// begins at block 1 and downstream consumers waiting for block 0 stall forever. Mirrors geth's -/// `OnGenesisBlock` hook: emitted with the full genesis alloc, only when starting on an empty -/// chain. -pub fn emit_genesis_block_if_empty(ctx: &ExExContext) -> eyre::Result<()> -where - Node: FullNodeComponents, - Node::Provider: BlockReader, - ChainSpec: EthereumHardforks + EthChainSpec, -{ - emit_genesis_block_on_empty_chain(ctx.provider(), ctx.config.chain.genesis()) -} - -/// Provider-based variant of [`emit_genesis_block_if_empty`] for embedders (op-reth, base) that -/// wire Firehose through their own engine validator instead of this crate's ExEx. Call it once -/// at node startup, after `on_blockchain_init` and before any block is traced. -pub fn emit_genesis_block_on_empty_chain

( - provider: &P, - genesis: &alloy_genesis::Genesis, -) -> eyre::Result<()> -where - P: BlockReader, - ::Header: BlockHeader + alloy_primitives::Sealable, - <::Body as reth_primitives_traits::BlockBody>::OmmerHeader: - BlockHeader + alloy_primitives::Sealable, -{ - use reth_primitives_traits::Block as _; - - // OP-stack chains may place their genesis at a non-zero height (e.g. OP mainnet's - // post-bedrock genesis); `genesis.number` carries it, defaulting to 0. - let genesis_number = genesis.number.unwrap_or(0); - if provider.best_block_number()? != genesis_number { - return Ok(()); - } - - let genesis_block = provider - .block(genesis_number.into())? - .ok_or_else(|| eyre::eyre!("genesis block {genesis_number} not found in database"))? - .seal_slow(); - - crate::tracer().on_genesis_block( - firehose_tracer::types::BlockEvent { - block: mapper::to_block_data(&genesis_block), - // Genesis is finalized by definition. - finalized: Some(firehose_tracer::types::FinalizedBlockRef { - number: genesis_number, - hash: Some(genesis_block.hash()), - }), - flash_block: None, - }, - mapper::to_genesis_alloc(genesis), - ); - - Ok(()) -}