use crate::{
builder::{NodeAdapter, NodeAddOns, NodeTypesAdapter},
components::{NodeComponents, NodeComponentsBuilder},
hooks::NodeHooks,
node::FullNode,
NodeBuilderWithComponents, NodeHandle,
};
use futures::{future::Either, stream, stream_select, StreamExt};
use reth_beacon_consensus::{
hooks::{EngineHooks, PruneHook, StaticFileHook},
BeaconConsensusEngine,
};
use reth_consensus_debug_client::{DebugConsensusClient, EtherscanBlockProvider, RpcBlockProvider};
use reth_engine_util::EngineMessageStreamExt;
use reth_exex::ExExManagerHandle;
use reth_network::NetworkEvents;
use reth_node_api::FullNodeTypes;
use reth_node_core::{
dirs::{ChainPath, DataDirPath},
exit::NodeExitFuture,
version::{CARGO_PKG_VERSION, CLIENT_CODE, NAME_CLIENT, VERGEN_GIT_SHA},
};
use reth_node_events::{cl::ConsensusLayerHealthEvents, node};
use reth_primitives::format_ether;
use reth_provider::providers::BlockchainProvider;
use reth_rpc_engine_api::EngineApi;
use reth_rpc_types::engine::ClientVersionV1;
use reth_tasks::TaskExecutor;
use reth_tracing::tracing::{debug, info};
use reth_transaction_pool::TransactionPool;
use std::{future::Future, sync::Arc};
use tokio::sync::{mpsc::unbounded_channel, oneshot};
use tokio_stream::wrappers::UnboundedReceiverStream;
pub mod common;
pub use common::LaunchContext;
mod exex;
pub use exex::ExExLauncher;
pub trait LaunchNode<Target> {
type Node;
fn launch_node(self, target: Target) -> impl Future<Output = eyre::Result<Self::Node>> + Send;
}
impl<F, Target, Fut, Node> LaunchNode<Target> for F
where
F: FnOnce(Target) -> Fut + Send,
Fut: Future<Output = eyre::Result<Node>> + Send,
{
type Node = Node;
fn launch_node(self, target: Target) -> impl Future<Output = eyre::Result<Self::Node>> + Send {
self(target)
}
}
#[derive(Debug)]
pub struct DefaultNodeLauncher {
pub ctx: LaunchContext,
}
impl DefaultNodeLauncher {
pub const fn new(task_executor: TaskExecutor, data_dir: ChainPath<DataDirPath>) -> Self {
Self { ctx: LaunchContext::new(task_executor, data_dir) }
}
}
impl<T, CB> LaunchNode<NodeBuilderWithComponents<T, CB>> for DefaultNodeLauncher
where
T: FullNodeTypes<Provider = BlockchainProvider<<T as FullNodeTypes>::DB>>,
CB: NodeComponentsBuilder<T>,
{
type Node = NodeHandle<NodeAdapter<T, CB::Components>>;
async fn launch_node(
self,
target: NodeBuilderWithComponents<T, CB>,
) -> eyre::Result<Self::Node> {
let Self { ctx } = self;
let NodeBuilderWithComponents {
adapter: NodeTypesAdapter { database },
components_builder,
add_ons: NodeAddOns { hooks, rpc, exexs: installed_exex },
config,
} = target;
let NodeHooks { on_component_initialized, on_node_started, .. } = hooks;
let ctx = ctx
.with_configured_globals()
.with_loaded_toml_config(config)?
.with_resolved_peers().await?
.attach(database.clone())
.with_adjusted_configs()
.with_provider_factory().await?
.inspect(|_| {
info!(target: "reth::cli", "Database opened");
})
.with_prometheus().await?
.inspect(|this| {
debug!(target: "reth::cli", chain=%this.chain_id(), genesis=?this.genesis_hash(), "Initializing genesis");
})
.with_genesis()?
.inspect(|this| {
info!(target: "reth::cli", "\n{}", this.chain_spec().display_hardforks());
})
.with_metrics()
.with_blockchain_db::<T>()?
.with_components(components_builder, on_component_initialized).await?;
let exex_manager_handle = ExExLauncher::new(
ctx.head(),
ctx.node_adapter().clone(),
installed_exex,
ctx.configs().clone(),
)
.launch()
.await;
let network_client = ctx.components().network().fetch_client().await?;
let (consensus_engine_tx, consensus_engine_rx) = unbounded_channel();
let node_config = ctx.node_config();
let consensus_engine_stream = UnboundedReceiverStream::from(consensus_engine_rx)
.maybe_skip_fcu(node_config.debug.skip_fcu)
.maybe_skip_new_payload(node_config.debug.skip_new_payload)
.maybe_store_messages(node_config.debug.engine_api_store.clone());
let max_block = ctx.max_block(network_client.clone()).await?;
let mut hooks = EngineHooks::new();
let static_file_producer = ctx.static_file_producer();
let static_file_producer_events = static_file_producer.lock().events();
hooks.add(StaticFileHook::new(
static_file_producer.clone(),
Box::new(ctx.task_executor().clone()),
));
info!(target: "reth::cli", "StaticFileProducer initialized");
let pipeline_exex_handle =
exex_manager_handle.clone().unwrap_or_else(ExExManagerHandle::empty);
let (pipeline, client) = if ctx.is_dev() {
info!(target: "reth::cli", "Starting Reth in dev mode");
for (idx, (address, alloc)) in ctx.chain_spec().genesis.alloc.iter().enumerate() {
info!(target: "reth::cli", "Allocated Genesis Account: {:02}. {} ({} ETH)", idx, address.to_string(), format_ether(alloc.balance));
}
let mining_mode =
ctx.dev_mining_mode(ctx.components().pool().pending_transactions_listener());
info!(target: "reth::cli", mode=%mining_mode, "configuring dev mining mode");
let (_, client, mut task) = reth_auto_seal_consensus::AutoSealBuilder::new(
ctx.chain_spec(),
ctx.blockchain_db().clone(),
ctx.components().pool().clone(),
consensus_engine_tx.clone(),
mining_mode,
ctx.components().block_executor().clone(),
)
.build();
let pipeline = crate::setup::build_networked_pipeline(
&ctx.toml_config().stages,
client.clone(),
ctx.consensus(),
ctx.provider_factory().clone(),
ctx.task_executor(),
ctx.sync_metrics_tx(),
ctx.prune_config(),
max_block,
static_file_producer,
ctx.components().block_executor().clone(),
pipeline_exex_handle,
)?;
let pipeline_events = pipeline.events();
task.set_pipeline_events(pipeline_events);
debug!(target: "reth::cli", "Spawning auto mine task");
ctx.task_executor().spawn(Box::pin(task));
(pipeline, Either::Left(client))
} else {
let pipeline = crate::setup::build_networked_pipeline(
&ctx.toml_config().stages,
network_client.clone(),
ctx.consensus(),
ctx.provider_factory().clone(),
ctx.task_executor(),
ctx.sync_metrics_tx(),
ctx.prune_config(),
max_block,
static_file_producer,
ctx.components().block_executor().clone(),
pipeline_exex_handle,
)?;
(pipeline, Either::Right(network_client.clone()))
};
let pipeline_events = pipeline.events();
let initial_target = ctx.node_config().debug.tip;
let mut pruner_builder =
ctx.pruner_builder().max_reorg_depth(ctx.tree_config().max_reorg_depth() as usize);
if let Some(exex_manager_handle) = &exex_manager_handle {
pruner_builder =
pruner_builder.finished_exex_height(exex_manager_handle.finished_height());
}
let pruner = pruner_builder.build(ctx.provider_factory().clone());
let pruner_events = pruner.events();
info!(target: "reth::cli", prune_config=?ctx.prune_config().unwrap_or_default(), "Pruner initialized");
hooks.add(PruneHook::new(pruner, Box::new(ctx.task_executor().clone())));
let (beacon_consensus_engine, beacon_engine_handle) = BeaconConsensusEngine::with_channel(
client,
pipeline,
ctx.blockchain_db().clone(),
Box::new(ctx.task_executor().clone()),
Box::new(ctx.components().network().clone()),
max_block,
ctx.components().payload_builder().clone(),
initial_target,
reth_beacon_consensus::MIN_BLOCKS_FOR_PIPELINE_RUN,
consensus_engine_tx,
Box::pin(consensus_engine_stream),
hooks,
)?;
info!(target: "reth::cli", "Consensus engine initialized");
let events = stream_select!(
ctx.components().network().event_listener().map(Into::into),
beacon_engine_handle.event_listener().map(Into::into),
pipeline_events.map(Into::into),
if ctx.node_config().debug.tip.is_none() && !ctx.is_dev() {
Either::Left(
ConsensusLayerHealthEvents::new(Box::new(ctx.blockchain_db().clone()))
.map(Into::into),
)
} else {
Either::Right(stream::empty())
},
pruner_events.map(Into::into),
static_file_producer_events.map(Into::into),
);
ctx.task_executor().spawn_critical(
"events task",
node::handle_events(
Some(ctx.components().network().clone()),
Some(ctx.head().number),
events,
database.clone(),
),
);
let client = ClientVersionV1 {
code: CLIENT_CODE,
name: NAME_CLIENT.to_string(),
version: CARGO_PKG_VERSION.to_string(),
commit: VERGEN_GIT_SHA.to_string(),
};
let engine_api = EngineApi::new(
ctx.blockchain_db().clone(),
ctx.chain_spec(),
beacon_engine_handle,
ctx.components().payload_builder().clone().into(),
Box::new(ctx.task_executor().clone()),
client,
);
info!(target: "reth::cli", "Engine API handler initialized");
let jwt_secret = ctx.auth_jwt_secret()?;
let (rpc_server_handles, rpc_registry) = crate::rpc::launch_rpc_servers(
ctx.node_adapter().clone(),
engine_api,
ctx.node_config(),
jwt_secret,
rpc,
)
.await?;
if ctx.is_dev() {
rpc_registry.eth_api().with_dev_accounts();
}
let (tx, rx) = oneshot::channel();
info!(target: "reth::cli", "Starting consensus engine");
ctx.task_executor().spawn_critical_blocking("consensus engine", async move {
let res = beacon_consensus_engine.await;
let _ = tx.send(res);
});
if let Some(maybe_custom_etherscan_url) = ctx.node_config().debug.etherscan.clone() {
info!(target: "reth::cli", "Using etherscan as consensus client");
let chain = ctx.node_config().chain.chain;
let etherscan_url = maybe_custom_etherscan_url.map(Ok).unwrap_or_else(|| {
chain
.etherscan_urls()
.map(|urls| urls.0.to_string())
.ok_or_else(|| eyre::eyre!("failed to get etherscan url for chain: {chain}"))
})?;
let block_provider = EtherscanBlockProvider::new(
etherscan_url,
chain.etherscan_api_key().ok_or_else(|| {
eyre::eyre!(
"etherscan api key not found for rpc consensus client for chain: {chain}"
)
})?,
);
let rpc_consensus_client = DebugConsensusClient::new(
rpc_server_handles.auth.clone(),
Arc::new(block_provider),
);
ctx.task_executor().spawn_critical("etherscan consensus client", async move {
rpc_consensus_client.run::<T::Engine>().await
});
}
if let Some(rpc_ws_url) = ctx.node_config().debug.rpc_consensus_ws.clone() {
info!(target: "reth::cli", "Using rpc provider as consensus client");
let block_provider = RpcBlockProvider::new(rpc_ws_url);
let rpc_consensus_client = DebugConsensusClient::new(
rpc_server_handles.auth.clone(),
Arc::new(block_provider),
);
ctx.task_executor().spawn_critical("rpc consensus client", async move {
rpc_consensus_client.run::<T::Engine>().await
});
}
let full_node = FullNode {
evm_config: ctx.components().evm_config().clone(),
block_executor: ctx.components().block_executor().clone(),
pool: ctx.components().pool().clone(),
network: ctx.components().network().clone(),
provider: ctx.node_adapter().provider.clone(),
payload_builder: ctx.components().payload_builder().clone(),
task_executor: ctx.task_executor().clone(),
rpc_server_handles,
rpc_registry,
config: ctx.node_config().clone(),
data_dir: ctx.data_dir().clone(),
};
on_node_started.on_event(full_node.clone())?;
let handle = NodeHandle {
node_exit_future: NodeExitFuture::new(
async { Ok(rx.await??) },
full_node.config.debug.terminate,
),
node: full_node,
};
Ok(handle)
}
}