Reth 存储层源码走读

Reth 是什么？

Reth is an Execution Layer (EL) and is compatible with all Ethereum Consensus Layer (CL) implementations that support the Engine API.

在以太坊 The Merge 之后，节点分成了两个部分，其中部分负责共识，另部分负责状态维护。

graph LR
    subgraph 共识层 CL
        cl1[Slot / Epoch 驱动]
        cl2[区块提议 / 投票]
        cl3[Finality 决定]
        cl4[Beacon 链同步]
    end

    subgraph 执行层 EL
        el1[交易执行（EVM）]
        el2[状态更新（State）]
        el3[交易池管理]
        el4[账户 / 区块存储]
    end

    cl1 -.-> API[Engine API]
    cl2 -.-> API
    cl3 -.-> API
    cl4 -.-> API
    API -.-> el1
    API -.-> el2
    API -.-> el3
    API -.-> el4

以太坊的状态

• 区块

  • 区块头
  • 区块体

• 链

  • 父叔
  • 高度

• 账户状态 [MPT]

  • balance

共识层通过 EngineAPI 调用执行层，那么 Engine API 有哪些语义？

• Paris
• Shanghai
• Cancun

为了实现 Engine API 的语义，执行层要有哪些能力？

Reth [执行层] = 逻辑层 + 存储层，存储层给逻辑层暴露了哪些能力？怎么样暴露的？

怎么样暴露的 ？

• Reth 中 BlockchainProvider 作为整个存储层的门户。

暴露了哪些能力 ？

• 实现了 数据结构的 Trait + 链状态的 Trait 【门面模式】，将存储的能力暴露给逻辑层。
  • Header
  • Block
  • Receipts
  • State
    • Latest
    • by_block_num
    • by_block_hash

#![allow(unused)]
fn main() {
crates/storage/provider/src/providers/blockchain_provider.rs

/// The main type for interacting with the blockchain.
///
/// This type serves as the main entry point for interacting with the blockchain and provides data
/// from database storage and from the blockchain tree (pending state etc.) It is a simple wrapper
/// type that holds an instance of the database and the blockchain tree.
#[derive(Debug)]
pub struct BlockchainProvider<N: NodeTypesWithDB> {
    /// Provider factory used to access the database.
    pub(crate) database: ProviderFactory<N>,
    /// Tracks the chain info wrt forkchoice updates and in memory canonical
    /// state.
    pub(crate) canonical_in_memory_state: CanonicalInMemoryState<N::Primitives>,
}
}

内存中管理的 Block

通过 canonical_in_memory_state 访问了哪些接口

BlockNumReader

#![allow(unused)]
fn main() {
impl<N: ProviderNodeTypes> BlockNumReader for BlockchainProvider<N> {
    fn chain_info(&self) -> ProviderResult<ChainInfo> {
        Ok(self.canonical_in_memory_state.chain_info())
    }

    fn best_block_number(&self) -> ProviderResult<BlockNumber> {
        Ok(self.canonical_in_memory_state.get_canonical_block_number())
    }

    fn last_block_number(&self) -> ProviderResult<BlockNumber> {
        self.database.last_block_number()
    }

    fn block_number(&self, hash: B256) -> ProviderResult<Option<BlockNumber>> {
        self.consistent_provider()?.block_number(hash)
    }
}
}

BlockIdReader

#![allow(unused)]
fn main() {
impl<N: ProviderNodeTypes> BlockIdReader for BlockchainProvider<N> {
    fn pending_block_num_hash(&self) -> ProviderResult<Option<BlockNumHash>> {
        Ok(self.canonical_in_memory_state.pending_block_num_hash())
    }

    fn safe_block_num_hash(&self) -> ProviderResult<Option<BlockNumHash>> {
        Ok(self.canonical_in_memory_state.get_safe_num_hash())
    }

    fn finalized_block_num_hash(&self) -> ProviderResult<Option<BlockNumHash>> {
        Ok(self.canonical_in_memory_state.get_finalized_num_hash())
    }
}
}

BlockReader

#![allow(unused)]
fn main() {
impl<N: ProviderNodeTypes> BlockReader for BlockchainProvider<N> {
    type Block = BlockTy<N>;

    fn find_block_by_hash(
        &self,
        hash: B256,
        source: BlockSource,
    ) -> ProviderResult<Option<Self::Block>> {
        self.consistent_provider()?.find_block_by_hash(hash, source)
    }

    fn block(&self, id: BlockHashOrNumber) -> ProviderResult<Option<Self::Block>> {
        self.consistent_provider()?.block(id)
    }

    fn pending_block(&self) -> ProviderResult<Option<SealedBlock<Self::Block>>> {
        Ok(self.canonical_in_memory_state.pending_block())
    }

    fn pending_block_with_senders(&self) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
        Ok(self.canonical_in_memory_state.pending_recovered_block())
    }

    fn pending_block_and_receipts(
        &self,
    ) -> ProviderResult<Option<(SealedBlock<Self::Block>, Vec<Self::Receipt>)>> {
        Ok(self.canonical_in_memory_state.pending_block_and_receipts())
    }
}

StateProviderFactory

#![allow(unused)]
fn main() {
impl<N: ProviderNodeTypes> StateProviderFactory for BlockchainProvider<N> {
    /// Storage provider for latest block
    fn latest(&self) -> ProviderResult<StateProviderBox> {
        trace!(target: "providers::blockchain", "Getting latest block state provider");
        // use latest state provider if the head state exists
        if let Some(state) = self.canonical_in_memory_state.head_state() {
            trace!(target: "providers::blockchain", "Using head state for latest state provider");
            Ok(self.block_state_provider(&state)?.boxed())
        } else {
            trace!(target: "providers::blockchain", "Using database state for latest state provider");
            self.database.latest()
  

    }
    /// Returns the state provider for pending state.
    ///
    /// If there's no pending block available then the latest state provider is returned:
    /// [`Self::latest`]
    fn pending(&self) -> ProviderResult<StateProviderBox> {
        trace!(target: "providers::blockchain", "Getting provider for pending state");

        if let Some(pending) = self.canonical_in_memory_state.pending_state() {
            // we have a pending block
            return Ok(Box::new(self.block_state_provider(&pending)?));
        }

        // fallback to latest state if the pending block is not available
        self.latest()
    }

    fn pending_state_by_hash(&self, block_hash: B256) -> ProviderResult<Option<StateProviderBox>> {
        if let Some(pending) = self.canonical_in_memory_state.pending_state() {
            if pending.hash() == block_hash {
                return Ok(Some(Box::new(self.block_state_provider(&pending)?)));
            }
        }
        Ok(None)
    }
}
}

CanonChainTracker

#![allow(unused)]
fn main() {
impl<N: ProviderNodeTypes> CanonChainTracker for BlockchainProvider<N> {
    type Header = HeaderTy<N>;

    fn on_forkchoice_update_received(&self, _update: &ForkchoiceState) {
        // update timestamp
        self.canonical_in_memory_state.on_forkchoice_update_received();
    }

    fn last_received_update_timestamp(&self) -> Option<Instant> {
        self.canonical_in_memory_state.last_received_update_timestamp()
    }

    fn set_canonical_head(&self, header: SealedHeader<Self::Header>) {
        self.canonical_in_memory_state.set_canonical_head(header);
    }

    fn set_safe(&self, header: SealedHeader<Self::Header>) {
        self.canonical_in_memory_state.set_safe(header);
    }

    fn set_finalized(&self, header: SealedHeader<Self::Header>) {
        self.canonical_in_memory_state.set_finalized(header);
    }
}
}

持久化层如何工作？

ProviderFactory 对外提供了哪些方法？

Reth 的架构鼓励通过 trait 来抽象每一种数据访问行为。ProviderFactory 实现了如下 trait 接口：

所有 trait 的背后实际调用的是 DatabaseProvider 对象，该对象则由 ProviderFactory 管理并创建。

典型工作场景

数据库层如何工作：数据分层

#![allow(unused)]
fn main() {
/// A provider struct that fetches data from the database.
/// Wrapper around [`DbTx`] and [`DbTxMut`]. Example: [`HeaderProvider`] [`BlockHashReader`]
#[derive(Debug)]
pub struct DatabaseProvider<TX, N: NodeTypes> {
    /// Database transaction.
    tx: TX,
    /// Chain spec
    chain_spec: Arc<N::ChainSpec>,
    /// Static File provider
    static_file_provider: StaticFileProvider<N::Primitives>,
    /// Pruning configuration
    prune_modes: PruneModes,
    /// Node storage handler.
    storage: Arc<N::Storage>,
}
}

• tx 提供数据库访问
• static_file_provider 提供静态文件访问

#![allow(unused)]
fn main() {
impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
    fn transactions_by_tx_range_with_cursor<C>(
        &self,
        range: impl RangeBounds<TxNumber>,
        cursor: &mut C,
    ) -> ProviderResult<Vec<TxTy<N>>>
    where
        C: DbCursorRO<tables::Transactions<TxTy<N>>>,
    {
        self.static_file_provider.get_range_with_static_file_or_database(
            StaticFileSegment::Transactions,
            to_range(range),
            |static_file, range, _| static_file.transactions_by_tx_range(range),
            |range, _| self.cursor_collect(cursor, range),
            |_| true,
        )
    }
}

• get_range_with_static_file_or_database

#![allow(unused)]
fn main() {
/// Gets data within a specified range, potentially spanning different `static_files` and_
/// database._
///_
/// # Arguments_
/// * `segment` - The segment of the static file to query._
/// * `block_range` - The range of data to fetch._
/// * `fetch_from_static_file` - A function to fetch data from the `static_file`._
/// * `fetch_from_database` - A function to fetch data from the database._
/// * `predicate` - A function used to evaluate each item in the fetched data. Fetching is_
///   terminated when this function returns false, thereby filtering the data based on the_
///   provided condition._
pub fn get_range_with_static_file_or_database<T, P, FS, FD>(
    &self,
    segment: StaticFileSegment,
    mut block_or_tx_range: Range<u64>,
    fetch_from_static_file: FS,
    mut fetch_from_database: FD,
    mut predicate: P,
) -> ProviderResult<Vec<T>>
where
    FS: Fn(&Self, Range<u64>, &mut P) -> ProviderResult<Vec<T>>,
    FD: FnMut(Range<u64>, P) -> ProviderResult<Vec<T>>,
    P: FnMut(&T) -> bool,
{
    let mut data = Vec::_new_();

    // If there is, check the maximum block or transaction number of the segment.
    if let _Some_(static_file_upper_bound) = if segment.is_block_based() {
        self.get_highest_static_file_block(segment)
    } else {
        self.get_highest_static_file_tx(segment)
    } {
        if block_or_tx_range.start <= static_file_upper_bound {
            let end = block_or_tx_range.end.min(static_file_upper_bound + 1);
            data.extend(fetch_from_static_file(
                self,
                block_or_tx_range.start..end,
                &mut predicate,
            )?);
            block_or_tx_range.start = end;
        }
    }

    if block_or_tx_range.end > block_or_tx_range.start {
        data.extend(fetch_from_database(block_or_tx_range, predicate)?)
    }

    _Ok_(data)
}
}

核心业务逻辑

1. 获取当前段的最大静态文件范围

#![allow(unused)]
fn main() {
let static_file_upper_bound = if segment.is_block_based() {
    self.get_highest_static_file_block(segment)
} else {
    self.get_highest_static_file_tx(segment)
};
}

• segment 可以基于区块或交易，函数会调用相应的方法来获取当前静态文件的最大已保存边界（例如最大区块号或最大交易号）。

2. 如果目标范围的一部分在静态文件中，就优先从静态文件读取

#![allow(unused)]
fn main() {
if block_or_tx_range.start <= static_file_upper_bound {
    let end = block_or_tx_range.end.min(static_file_upper_bound + 1);
    data.extend(fetch_from_static_file(
        self,
        block_or_tx_range.start..end,
        &mut predicate,
    )?);
    block_or_tx_range.start = end;
}
}

• 如果目标范围的起始值小于等于静态文件边界，说明这部分数据可以从静态文件中读取。
• 读取的实际区间是 [start, min(end, static_file_upper_bound + 1))
• 调用 fetch_from_static_file，并传入可变谓词 &mut predicate 用于过滤。
• 然后更新 block_or_tx_range.start = end，只保留未处理的区间。

3. 对于剩下的区间（即静态文件中没有的），从数据库读取

#![allow(unused)]
fn main() {
if block_or_tx_range.end > block_or_tx_range.start {
    data.extend(fetch_from_database(block_or_tx_range, predicate)?)
}
}

• 如果仍有未处理的区间（即 end > start），则调用 fetch_from_database 获取数据。

4. 返回所有结果

#![allow(unused)]
fn main() {
Ok(data)
}