reth_optimism_cli/commands/
import.rs

1//! Command that initializes the node by importing OP Mainnet chain segment below Bedrock, from a
2//! file.
3use 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::version_metadata;
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/// Syncs RLP encoded blocks from a file.
25#[derive(Debug, Parser)]
26pub struct ImportOpCommand<C: ChainSpecParser> {
27    #[command(flatten)]
28    env: EnvironmentArgs<C>,
29
30    /// Chunk byte length to read from file.
31    #[arg(long, value_name = "CHUNK_LEN", verbatim_doc_comment)]
32    chunk_len: Option<u64>,
33
34    /// The path to a block file for import.
35    ///
36    /// The online stages (headers and bodies) are replaced by a file import, after which the
37    /// remaining stages are executed.
38    #[arg(value_name = "IMPORT_PATH", verbatim_doc_comment)]
39    path: PathBuf,
40}
41
42impl<C: ChainSpecParser<ChainSpec = OpChainSpec>> ImportOpCommand<C> {
43    /// Execute `import` command
44    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec, Primitives = OpPrimitives>>(
45        self,
46    ) -> eyre::Result<()> {
47        info!(target: "reth::cli", "reth {} starting", version_metadata().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        // we use noop here because we expect the inputs to be valid
61        let consensus = Arc::new(NoopConsensus::default());
62
63        // open file
64        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        let static_file_producer =
75            StaticFileProducer::new(provider_factory.clone(), PruneModes::default());
76
77        while let Some(mut file_client) =
78            reader.next_chunk::<BlockTy<N>>(consensus.clone(), Some(sealed_header)).await?
79        {
80            // create a new FileClient from chunk read from file
81            info!(target: "reth::cli",
82                "Importing chain file chunk"
83            );
84
85            let tip = file_client.tip().ok_or_else(|| eyre::eyre!("file client has no tip"))?;
86            info!(target: "reth::cli", "Chain file chunk read");
87
88            total_decoded_blocks += file_client.headers_len();
89            total_decoded_txns += file_client.total_transactions();
90
91            for (block_number, body) in file_client.bodies_iter_mut() {
92                body.transactions.retain(|_| {
93                    if is_dup_tx(block_number) {
94                        total_filtered_out_dup_txns += 1;
95                        return false
96                    }
97                    true
98                })
99            }
100
101            let (mut pipeline, events) = build_import_pipeline(
102                &config,
103                provider_factory.clone(),
104                &consensus,
105                Arc::new(file_client),
106                static_file_producer.clone(),
107                true,
108                OpExecutorProvider::optimism(provider_factory.chain_spec()),
109            )?;
110
111            // override the tip
112            pipeline.set_tip(tip);
113            debug!(target: "reth::cli", ?tip, "Tip manually set");
114
115            let provider = provider_factory.provider()?;
116
117            let latest_block_number =
118                provider.get_stage_checkpoint(StageId::Finish)?.map(|ch| ch.block_number);
119            tokio::spawn(reth_node_events::node::handle_events(None, latest_block_number, events));
120
121            // Run pipeline
122            info!(target: "reth::cli", "Starting sync pipeline");
123            tokio::select! {
124                res = pipeline.run() => res?,
125                _ = tokio::signal::ctrl_c() => {},
126            }
127
128            sealed_header = provider_factory
129                .sealed_header(provider_factory.last_block_number()?)?
130                .expect("should have genesis");
131        }
132
133        let provider = provider_factory.provider()?;
134
135        let total_imported_blocks = provider.tx_ref().entries::<tables::HeaderNumbers>()?;
136        let total_imported_txns = provider.tx_ref().entries::<tables::TransactionHashNumbers>()?;
137
138        if total_decoded_blocks != total_imported_blocks ||
139            total_decoded_txns != total_imported_txns + total_filtered_out_dup_txns
140        {
141            error!(target: "reth::cli",
142                total_decoded_blocks,
143                total_imported_blocks,
144                total_decoded_txns,
145                total_filtered_out_dup_txns,
146                total_imported_txns,
147                "Chain was partially imported"
148            );
149        }
150
151        info!(target: "reth::cli",
152            total_imported_blocks,
153            total_imported_txns,
154            total_decoded_blocks,
155            total_decoded_txns,
156            total_filtered_out_dup_txns,
157            "Chain file imported"
158        );
159
160        Ok(())
161    }
162}
163
164impl<C: ChainSpecParser> ImportOpCommand<C> {
165    /// Returns the underlying chain being used to run this command
166    pub const fn chain_spec(&self) -> Option<&Arc<C::ChainSpec>> {
167        Some(&self.env.chain)
168    }
169}
reth_optimism_cli/commands/import.rs

reth_optimism_cli/commands/
import.rs
