use std::sync::Arc;
use reth_basic_payload_builder::{BasicPayloadJobGenerator, BasicPayloadJobGeneratorConfig};
use reth_chainspec::{EthChainSpec, Hardforks};
use reth_evm::{execute::BasicBlockExecutorProvider, ConfigureEvm};
use reth_network::{NetworkConfig, NetworkHandle, NetworkManager, PeersInfo};
use reth_node_api::{
AddOnsContext, EngineValidator, FullNodeComponents, NodeAddOns, NodePrimitives,
};
use reth_node_builder::{
components::{
ComponentsBuilder, ConsensusBuilder, ExecutorBuilder, NetworkBuilder,
PayloadServiceBuilder, PoolBuilder, PoolBuilderConfigOverrides,
},
node::{FullNodeTypes, NodeTypes, NodeTypesWithEngine},
rpc::{EngineValidatorBuilder, RethRpcAddOns, RpcAddOns, RpcHandle},
BuilderContext, Node, NodeAdapter, NodeComponentsBuilder, PayloadBuilderConfig,
};
use reth_optimism_chainspec::OpChainSpec;
use reth_optimism_consensus::OpBeaconConsensus;
use reth_optimism_evm::{OpEvmConfig, OpExecutionStrategyFactory};
use reth_optimism_payload_builder::builder::OpPayloadTransactions;
use reth_optimism_rpc::OpEthApi;
use reth_payload_builder::{PayloadBuilderHandle, PayloadBuilderService};
use reth_primitives::{Block, Header};
use reth_provider::CanonStateSubscriptions;
use reth_tracing::tracing::{debug, info};
use reth_transaction_pool::{
blobstore::DiskFileBlobStore, CoinbaseTipOrdering, TransactionPool,
TransactionValidationTaskExecutor,
};
use reth_trie_db::MerklePatriciaTrie;
use crate::{
args::RollupArgs,
engine::OpEngineValidator,
txpool::{OpTransactionPool, OpTransactionValidator},
OpEngineTypes,
};
#[derive(Debug)]
pub struct OpPrimitives;
impl NodePrimitives for OpPrimitives {
type Block = Block;
}
#[derive(Debug, Default, Clone)]
#[non_exhaustive]
pub struct OptimismNode {
pub args: RollupArgs,
}
impl OptimismNode {
pub const fn new(args: RollupArgs) -> Self {
Self { args }
}
pub fn components<Node>(
args: RollupArgs,
) -> ComponentsBuilder<
Node,
OpPoolBuilder,
OpPayloadBuilder,
OpNetworkBuilder,
OpExecutorBuilder,
OpConsensusBuilder,
>
where
Node: FullNodeTypes<
Types: NodeTypesWithEngine<Engine = OpEngineTypes, ChainSpec = OpChainSpec>,
>,
{
let RollupArgs { disable_txpool_gossip, compute_pending_block, discovery_v4, .. } = args;
ComponentsBuilder::default()
.node_types::<Node>()
.pool(OpPoolBuilder::default())
.payload(OpPayloadBuilder::new(compute_pending_block))
.network(OpNetworkBuilder {
disable_txpool_gossip,
disable_discovery_v4: !discovery_v4,
})
.executor(OpExecutorBuilder::default())
.consensus(OpConsensusBuilder::default())
}
}
impl<N> Node<N> for OptimismNode
where
N: FullNodeTypes<Types: NodeTypesWithEngine<Engine = OpEngineTypes, ChainSpec = OpChainSpec>>,
{
type ComponentsBuilder = ComponentsBuilder<
N,
OpPoolBuilder,
OpPayloadBuilder,
OpNetworkBuilder,
OpExecutorBuilder,
OpConsensusBuilder,
>;
type AddOns = OptimismAddOns<
NodeAdapter<N, <Self::ComponentsBuilder as NodeComponentsBuilder<N>>::Components>,
>;
fn components_builder(&self) -> Self::ComponentsBuilder {
let Self { args } = self;
Self::components(args.clone())
}
fn add_ons(&self) -> Self::AddOns {
OptimismAddOns::new(self.args.sequencer_http.clone())
}
}
impl NodeTypes for OptimismNode {
type Primitives = OpPrimitives;
type ChainSpec = OpChainSpec;
type StateCommitment = MerklePatriciaTrie;
}
impl NodeTypesWithEngine for OptimismNode {
type Engine = OpEngineTypes;
}
#[derive(Debug)]
pub struct OptimismAddOns<N: FullNodeComponents>(
pub RpcAddOns<N, OpEthApi<N>, OptimismEngineValidatorBuilder>,
);
impl<N: FullNodeComponents> Default for OptimismAddOns<N> {
fn default() -> Self {
Self::new(None)
}
}
impl<N: FullNodeComponents> OptimismAddOns<N> {
pub fn new(sequencer_http: Option<String>) -> Self {
Self(RpcAddOns::new(move |ctx| OpEthApi::new(ctx, sequencer_http), Default::default()))
}
}
impl<N> NodeAddOns<N> for OptimismAddOns<N>
where
N: FullNodeComponents<Types: NodeTypes<ChainSpec = OpChainSpec>>,
OpEngineValidator: EngineValidator<<N::Types as NodeTypesWithEngine>::Engine>,
{
type Handle = RpcHandle<N, OpEthApi<N>>;
async fn launch_add_ons(
self,
ctx: reth_node_api::AddOnsContext<'_, N>,
) -> eyre::Result<Self::Handle> {
self.0.launch_add_ons(ctx).await
}
}
impl<N> RethRpcAddOns<N> for OptimismAddOns<N>
where
N: FullNodeComponents<Types: NodeTypes<ChainSpec = OpChainSpec>>,
OpEngineValidator: EngineValidator<<N::Types as NodeTypesWithEngine>::Engine>,
{
type EthApi = OpEthApi<N>;
fn hooks_mut(&mut self) -> &mut reth_node_builder::rpc::RpcHooks<N, Self::EthApi> {
self.0.hooks_mut()
}
}
#[derive(Debug, Default, Clone, Copy)]
#[non_exhaustive]
pub struct OpExecutorBuilder;
impl<Node> ExecutorBuilder<Node> for OpExecutorBuilder
where
Node: FullNodeTypes<Types: NodeTypes<ChainSpec = OpChainSpec>>,
{
type EVM = OpEvmConfig;
type Executor = BasicBlockExecutorProvider<OpExecutionStrategyFactory>;
async fn build_evm(
self,
ctx: &BuilderContext<Node>,
) -> eyre::Result<(Self::EVM, Self::Executor)> {
let evm_config = OpEvmConfig::new(ctx.chain_spec());
let strategy_factory =
OpExecutionStrategyFactory::new(ctx.chain_spec(), evm_config.clone());
let executor = BasicBlockExecutorProvider::new(strategy_factory);
Ok((evm_config, executor))
}
}
#[derive(Debug, Default, Clone)]
pub struct OpPoolBuilder {
pub pool_config_overrides: PoolBuilderConfigOverrides,
}
impl<Node> PoolBuilder<Node> for OpPoolBuilder
where
Node: FullNodeTypes<Types: NodeTypes<ChainSpec = OpChainSpec>>,
{
type Pool = OpTransactionPool<Node::Provider, DiskFileBlobStore>;
async fn build_pool(self, ctx: &BuilderContext<Node>) -> eyre::Result<Self::Pool> {
let Self { pool_config_overrides } = self;
let data_dir = ctx.config().datadir();
let blob_store = DiskFileBlobStore::open(data_dir.blobstore(), Default::default())?;
let validator = TransactionValidationTaskExecutor::eth_builder(Arc::new(
ctx.chain_spec().inner.clone(),
))
.with_head_timestamp(ctx.head().timestamp)
.kzg_settings(ctx.kzg_settings()?)
.with_additional_tasks(
pool_config_overrides
.additional_validation_tasks
.unwrap_or_else(|| ctx.config().txpool.additional_validation_tasks),
)
.build_with_tasks(ctx.provider().clone(), ctx.task_executor().clone(), blob_store.clone())
.map(|validator| {
OpTransactionValidator::new(validator)
.require_l1_data_gas_fee(!ctx.config().dev.dev)
});
let transaction_pool = reth_transaction_pool::Pool::new(
validator,
CoinbaseTipOrdering::default(),
blob_store,
pool_config_overrides.apply(ctx.pool_config()),
);
info!(target: "reth::cli", "Transaction pool initialized");
let transactions_path = data_dir.txpool_transactions();
{
let pool = transaction_pool.clone();
let chain_events = ctx.provider().canonical_state_stream();
let client = ctx.provider().clone();
let transactions_backup_config =
reth_transaction_pool::maintain::LocalTransactionBackupConfig::with_local_txs_backup(transactions_path);
ctx.task_executor().spawn_critical_with_graceful_shutdown_signal(
"local transactions backup task",
|shutdown| {
reth_transaction_pool::maintain::backup_local_transactions_task(
shutdown,
pool.clone(),
transactions_backup_config,
)
},
);
ctx.task_executor().spawn_critical(
"txpool maintenance task",
reth_transaction_pool::maintain::maintain_transaction_pool_future(
client,
pool,
chain_events,
ctx.task_executor().clone(),
Default::default(),
),
);
debug!(target: "reth::cli", "Spawned txpool maintenance task");
}
Ok(transaction_pool)
}
}
#[derive(Debug, Default, Clone)]
pub struct OpPayloadBuilder<Txs = ()> {
pub compute_pending_block: bool,
pub best_transactions: Txs,
}
impl OpPayloadBuilder {
pub const fn new(compute_pending_block: bool) -> Self {
Self { compute_pending_block, best_transactions: () }
}
}
impl<Txs> OpPayloadBuilder<Txs>
where
Txs: OpPayloadTransactions,
{
pub fn with_transactions<T: OpPayloadTransactions>(
self,
best_transactions: T,
) -> OpPayloadBuilder<T> {
let Self { compute_pending_block, .. } = self;
OpPayloadBuilder { compute_pending_block, best_transactions }
}
pub fn spawn<Node, Evm, Pool>(
self,
evm_config: Evm,
ctx: &BuilderContext<Node>,
pool: Pool,
) -> eyre::Result<PayloadBuilderHandle<OpEngineTypes>>
where
Node: FullNodeTypes<
Types: NodeTypesWithEngine<Engine = OpEngineTypes, ChainSpec = OpChainSpec>,
>,
Pool: TransactionPool + Unpin + 'static,
Evm: ConfigureEvm<Header = Header>,
{
let payload_builder = reth_optimism_payload_builder::OpPayloadBuilder::new(evm_config)
.with_transactions(self.best_transactions)
.set_compute_pending_block(self.compute_pending_block);
let conf = ctx.payload_builder_config();
let payload_job_config = BasicPayloadJobGeneratorConfig::default()
.interval(conf.interval())
.deadline(conf.deadline())
.max_payload_tasks(conf.max_payload_tasks())
.extradata(Default::default());
let payload_generator = BasicPayloadJobGenerator::with_builder(
ctx.provider().clone(),
pool,
ctx.task_executor().clone(),
payload_job_config,
payload_builder,
);
let (payload_service, payload_builder) =
PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
ctx.task_executor().spawn_critical("payload builder service", Box::pin(payload_service));
Ok(payload_builder)
}
}
impl<Node, Pool, Txs> PayloadServiceBuilder<Node, Pool> for OpPayloadBuilder<Txs>
where
Node:
FullNodeTypes<Types: NodeTypesWithEngine<Engine = OpEngineTypes, ChainSpec = OpChainSpec>>,
Pool: TransactionPool + Unpin + 'static,
Txs: OpPayloadTransactions,
{
async fn spawn_payload_service(
self,
ctx: &BuilderContext<Node>,
pool: Pool,
) -> eyre::Result<PayloadBuilderHandle<OpEngineTypes>> {
self.spawn(OpEvmConfig::new(ctx.chain_spec()), ctx, pool)
}
}
#[derive(Debug, Default, Clone)]
pub struct OpNetworkBuilder {
pub disable_txpool_gossip: bool,
pub disable_discovery_v4: bool,
}
impl OpNetworkBuilder {
pub fn network_config<Node>(
&self,
ctx: &BuilderContext<Node>,
) -> eyre::Result<NetworkConfig<<Node as FullNodeTypes>::Provider>>
where
Node: FullNodeTypes<Types: NodeTypes<ChainSpec: Hardforks>>,
{
let Self { disable_txpool_gossip, disable_discovery_v4 } = self.clone();
let args = &ctx.config().network;
let network_builder = ctx
.network_config_builder()?
.apply(|mut builder| {
let rlpx_socket = (args.addr, args.port).into();
if disable_discovery_v4 || args.discovery.disable_discovery {
builder = builder.disable_discv4_discovery();
}
if !args.discovery.disable_discovery {
builder = builder.discovery_v5(
args.discovery.discovery_v5_builder(
rlpx_socket,
ctx.config()
.network
.resolved_bootnodes()
.or_else(|| ctx.chain_spec().bootnodes())
.unwrap_or_default(),
),
);
}
builder
});
let mut network_config = ctx.build_network_config(network_builder);
network_config.tx_gossip_disabled = disable_txpool_gossip;
Ok(network_config)
}
}
impl<Node, Pool> NetworkBuilder<Node, Pool> for OpNetworkBuilder
where
Node: FullNodeTypes<Types: NodeTypes<ChainSpec = OpChainSpec>>,
Pool: TransactionPool + Unpin + 'static,
{
async fn build_network(
self,
ctx: &BuilderContext<Node>,
pool: Pool,
) -> eyre::Result<NetworkHandle> {
let network_config = self.network_config(ctx)?;
let network = NetworkManager::builder(network_config).await?;
let handle = ctx.start_network(network, pool);
info!(target: "reth::cli", enode=%handle.local_node_record(), "P2P networking initialized");
Ok(handle)
}
}
#[derive(Debug, Default, Clone)]
#[non_exhaustive]
pub struct OpConsensusBuilder;
impl<Node> ConsensusBuilder<Node> for OpConsensusBuilder
where
Node: FullNodeTypes<Types: NodeTypes<ChainSpec = OpChainSpec>>,
{
type Consensus = Arc<dyn reth_consensus::Consensus>;
async fn build_consensus(self, ctx: &BuilderContext<Node>) -> eyre::Result<Self::Consensus> {
if ctx.is_dev() {
Ok(Arc::new(reth_auto_seal_consensus::AutoSealConsensus::new(ctx.chain_spec())))
} else {
Ok(Arc::new(OpBeaconConsensus::new(ctx.chain_spec())))
}
}
}
#[derive(Debug, Default, Clone)]
#[non_exhaustive]
pub struct OptimismEngineValidatorBuilder;
impl<Node, Types> EngineValidatorBuilder<Node> for OptimismEngineValidatorBuilder
where
Types: NodeTypesWithEngine<ChainSpec = OpChainSpec>,
Node: FullNodeComponents<Types = Types>,
OpEngineValidator: EngineValidator<Types::Engine>,
{
type Validator = OpEngineValidator;
async fn build(self, ctx: &AddOnsContext<'_, Node>) -> eyre::Result<Self::Validator> {
Ok(OpEngineValidator::new(ctx.config.chain.clone()))
}
}