reth_optimism_cli/commands/
import.rs
1use clap::Parser;
4use reth_cli::chainspec::ChainSpecParser;
5use reth_cli_commands::{
6 common::{AccessRights, CliNodeTypes, Environment, EnvironmentArgs},
7 import::build_import_pipeline,
8};
9use reth_consensus::noop::NoopConsensus;
10use reth_db_api::{tables, transaction::DbTx};
11use reth_downloaders::file_client::{ChunkedFileReader, DEFAULT_BYTE_LEN_CHUNK_CHAIN_FILE};
12use reth_node_builder::BlockTy;
13use reth_node_core::version::SHORT_VERSION;
14use reth_optimism_chainspec::OpChainSpec;
15use reth_optimism_evm::OpExecutorProvider;
16use reth_optimism_primitives::{bedrock::is_dup_tx, OpPrimitives};
17use reth_provider::{BlockNumReader, ChainSpecProvider, HeaderProvider, StageCheckpointReader};
18use reth_prune::PruneModes;
19use reth_stages::StageId;
20use reth_static_file::StaticFileProducer;
21use std::{path::PathBuf, sync::Arc};
22use tracing::{debug, error, info};
23
24#[derive(Debug, Parser)]
26pub struct ImportOpCommand<C: ChainSpecParser> {
27 #[command(flatten)]
28 env: EnvironmentArgs<C>,
29
30 #[arg(long, value_name = "CHUNK_LEN", verbatim_doc_comment)]
32 chunk_len: Option<u64>,
33
34 #[arg(value_name = "IMPORT_PATH", verbatim_doc_comment)]
39 path: PathBuf,
40}
41
42impl<C: ChainSpecParser<ChainSpec = OpChainSpec>> ImportOpCommand<C> {
43 pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec, Primitives = OpPrimitives>>(
45 self,
46 ) -> eyre::Result<()> {
47 info!(target: "reth::cli", "reth {} starting", SHORT_VERSION);
48
49 info!(target: "reth::cli",
50 "Disabled stages requiring state, since cannot execute OVM state changes"
51 );
52
53 debug!(target: "reth::cli",
54 chunk_byte_len=self.chunk_len.unwrap_or(DEFAULT_BYTE_LEN_CHUNK_CHAIN_FILE),
55 "Chunking chain import"
56 );
57
58 let Environment { provider_factory, config, .. } = self.env.init::<N>(AccessRights::RW)?;
59
60 let consensus = Arc::new(NoopConsensus::default());
62
63 let mut reader = ChunkedFileReader::new(&self.path, self.chunk_len).await?;
65
66 let mut total_decoded_blocks = 0;
67 let mut total_decoded_txns = 0;
68 let mut total_filtered_out_dup_txns = 0;
69
70 let mut sealed_header = provider_factory
71 .sealed_header(provider_factory.last_block_number()?)?
72 .expect("should have genesis");
73
74 while let Some(mut file_client) =
75 reader.next_chunk::<BlockTy<N>>(consensus.clone(), Some(sealed_header)).await?
76 {
77 info!(target: "reth::cli",
79 "Importing chain file chunk"
80 );
81
82 let tip = file_client.tip().ok_or_else(|| eyre::eyre!("file client has no tip"))?;
83 info!(target: "reth::cli", "Chain file chunk read");
84
85 total_decoded_blocks += file_client.headers_len();
86 total_decoded_txns += file_client.total_transactions();
87
88 for (block_number, body) in file_client.bodies_iter_mut() {
89 body.transactions.retain(|_| {
90 if is_dup_tx(block_number) {
91 total_filtered_out_dup_txns += 1;
92 return false
93 }
94 true
95 })
96 }
97
98 let (mut pipeline, events) = build_import_pipeline(
99 &config,
100 provider_factory.clone(),
101 &consensus,
102 Arc::new(file_client),
103 StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
104 true,
105 OpExecutorProvider::optimism(provider_factory.chain_spec()),
106 )?;
107
108 pipeline.set_tip(tip);
110 debug!(target: "reth::cli", ?tip, "Tip manually set");
111
112 let provider = provider_factory.provider()?;
113
114 let latest_block_number =
115 provider.get_stage_checkpoint(StageId::Finish)?.map(|ch| ch.block_number);
116 tokio::spawn(reth_node_events::node::handle_events(None, latest_block_number, events));
117
118 info!(target: "reth::cli", "Starting sync pipeline");
120 tokio::select! {
121 res = pipeline.run() => res?,
122 _ = tokio::signal::ctrl_c() => {},
123 }
124
125 sealed_header = provider_factory
126 .sealed_header(provider_factory.last_block_number()?)?
127 .expect("should have genesis");
128 }
129
130 let provider = provider_factory.provider()?;
131
132 let total_imported_blocks = provider.tx_ref().entries::<tables::HeaderNumbers>()?;
133 let total_imported_txns = provider.tx_ref().entries::<tables::TransactionHashNumbers>()?;
134
135 if total_decoded_blocks != total_imported_blocks ||
136 total_decoded_txns != total_imported_txns + total_filtered_out_dup_txns
137 {
138 error!(target: "reth::cli",
139 total_decoded_blocks,
140 total_imported_blocks,
141 total_decoded_txns,
142 total_filtered_out_dup_txns,
143 total_imported_txns,
144 "Chain was partially imported"
145 );
146 }
147
148 info!(target: "reth::cli",
149 total_imported_blocks,
150 total_imported_txns,
151 total_decoded_blocks,
152 total_decoded_txns,
153 total_filtered_out_dup_txns,
154 "Chain file imported"
155 );
156
157 Ok(())
158 }
159}