|
| 1 | +use alloy_primitives::{BlockHash, BlockNumber}; |
| 2 | +use futures_util::StreamExt; |
| 3 | +use reth_db_api::{ |
| 4 | + cursor::{DbCursorRO, DbCursorRW}, |
| 5 | + table::Value, |
| 6 | + tables, |
| 7 | + transaction::{DbTx, DbTxMut}, |
| 8 | + RawKey, RawTable, RawValue, |
| 9 | +}; |
| 10 | +use reth_era::{era1_file::Era1Reader, execution_types::DecodeCompressed}; |
| 11 | +use reth_era_downloader::{EraStream, HttpClient}; |
| 12 | +use reth_etl::Collector; |
| 13 | +use reth_fs_util as fs; |
| 14 | +use reth_primitives_traits::{Block, FullBlockBody, FullBlockHeader, NodePrimitives}; |
| 15 | +use reth_provider::{ |
| 16 | + BlockWriter, ProviderError, StaticFileProviderFactory, StaticFileSegment, StaticFileWriter, |
| 17 | +}; |
| 18 | +use reth_storage_api::{DBProvider, HeaderProvider, NodePrimitivesProvider, StorageLocation}; |
| 19 | +use std::sync::mpsc; |
| 20 | +use tracing::info; |
| 21 | + |
| 22 | +/// Imports blocks from `downloader` using `provider`. |
| 23 | +/// |
| 24 | +/// Returns current block height. |
| 25 | +pub fn import<H, P, B, BB, BH>( |
| 26 | + mut downloader: EraStream<H>, |
| 27 | + provider: &P, |
| 28 | + mut hash_collector: Collector<BlockHash, BlockNumber>, |
| 29 | +) -> eyre::Result<BlockNumber> |
| 30 | +where |
| 31 | + B: Block<Header = BH, Body = BB>, |
| 32 | + BH: FullBlockHeader + Value, |
| 33 | + BB: FullBlockBody< |
| 34 | + Transaction = <<P as NodePrimitivesProvider>::Primitives as NodePrimitives>::SignedTx, |
| 35 | + OmmerHeader = BH, |
| 36 | + >, |
| 37 | + H: HttpClient + Clone + Send + Sync + 'static + Unpin, |
| 38 | + P: DBProvider<Tx: DbTxMut> + StaticFileProviderFactory + BlockWriter<Block = B>, |
| 39 | + <P as NodePrimitivesProvider>::Primitives: NodePrimitives<BlockHeader = BH, BlockBody = BB>, |
| 40 | +{ |
| 41 | + let (tx, rx) = mpsc::channel(); |
| 42 | + |
| 43 | + // Handle IO-bound async download in a background tokio task |
| 44 | + tokio::spawn(async move { |
| 45 | + while let Some(file) = downloader.next().await { |
| 46 | + tx.send(Some(file))?; |
| 47 | + } |
| 48 | + tx.send(None) |
| 49 | + }); |
| 50 | + |
| 51 | + let static_file_provider = provider.static_file_provider(); |
| 52 | + |
| 53 | + // Consistency check of expected headers in static files vs DB is done on provider::sync_gap |
| 54 | + // when poll_execute_ready is polled. |
| 55 | + let mut last_header_number = static_file_provider |
| 56 | + .get_highest_static_file_block(StaticFileSegment::Headers) |
| 57 | + .unwrap_or_default(); |
| 58 | + |
| 59 | + // Find the latest total difficulty |
| 60 | + let mut td = static_file_provider |
| 61 | + .header_td_by_number(last_header_number)? |
| 62 | + .ok_or(ProviderError::TotalDifficultyNotFound(last_header_number))?; |
| 63 | + |
| 64 | + // Although headers were downloaded in reverse order, the collector iterates it in ascending |
| 65 | + // order |
| 66 | + let mut writer = static_file_provider.latest_writer(StaticFileSegment::Headers)?; |
| 67 | + |
| 68 | + while let Some(path) = rx.recv()? { |
| 69 | + let path = path?; |
| 70 | + let file = fs::open(path.clone())?; |
| 71 | + let mut reader = Era1Reader::new(file); |
| 72 | + |
| 73 | + for block in reader.iter() { |
| 74 | + let block = block?; |
| 75 | + let header: BH = block.header.decode()?; |
| 76 | + let body: BB = block.body.decode()?; |
| 77 | + let number = header.number(); |
| 78 | + |
| 79 | + if number == 0 { |
| 80 | + continue; |
| 81 | + } |
| 82 | + |
| 83 | + let hash = header.hash_slow(); |
| 84 | + last_header_number = number; |
| 85 | + |
| 86 | + // Increase total difficulty |
| 87 | + td += header.difficulty(); |
| 88 | + |
| 89 | + // Append to Headers segment |
| 90 | + writer.append_header(&header, td, &hash)?; |
| 91 | + |
| 92 | + // Write bodies to database. |
| 93 | + provider.append_block_bodies( |
| 94 | + vec![(header.number(), Some(body))], |
| 95 | + // We are writing transactions directly to static files. |
| 96 | + StorageLocation::StaticFiles, |
| 97 | + )?; |
| 98 | + |
| 99 | + hash_collector.insert(hash, number)?; |
| 100 | + } |
| 101 | + |
| 102 | + info!(target: "era::history::import", "Processed {}", path.to_string_lossy()); |
| 103 | + |
| 104 | + fs::remove_file(path)?; |
| 105 | + } |
| 106 | + |
| 107 | + let total_headers = hash_collector.len(); |
| 108 | + info!(target: "era::history::import", total = total_headers, "Writing headers hash index"); |
| 109 | + |
| 110 | + // Database cursor for hash to number index |
| 111 | + let mut cursor_header_numbers = |
| 112 | + provider.tx_ref().cursor_write::<RawTable<tables::HeaderNumbers>>()?; |
| 113 | + let mut first_sync = false; |
| 114 | + |
| 115 | + // If we only have the genesis block hash, then we are at first sync, and we can remove it, |
| 116 | + // add it to the collector and use tx.append on all hashes. |
| 117 | + if provider.tx_ref().entries::<RawTable<tables::HeaderNumbers>>()? == 1 { |
| 118 | + if let Some((hash, block_number)) = cursor_header_numbers.last()? { |
| 119 | + if block_number.value()? == 0 { |
| 120 | + hash_collector.insert(hash.key()?, 0)?; |
| 121 | + cursor_header_numbers.delete_current()?; |
| 122 | + first_sync = true; |
| 123 | + } |
| 124 | + } |
| 125 | + } |
| 126 | + |
| 127 | + let interval = (total_headers / 10).max(1); |
| 128 | + |
| 129 | + // Build block hash to block number index |
| 130 | + for (index, hash_to_number) in hash_collector.iter()?.enumerate() { |
| 131 | + let (hash, number) = hash_to_number?; |
| 132 | + |
| 133 | + if index > 0 && index % interval == 0 && total_headers > 100 { |
| 134 | + info!(target: "era::history::import", progress = %format!("{:.2}%", (index as f64 / total_headers as f64) * 100.0), "Writing headers hash index"); |
| 135 | + } |
| 136 | + |
| 137 | + let hash = RawKey::<BlockHash>::from_vec(hash); |
| 138 | + let number = RawValue::<BlockNumber>::from_vec(number); |
| 139 | + |
| 140 | + if first_sync { |
| 141 | + cursor_header_numbers.append(hash, &number)?; |
| 142 | + } else { |
| 143 | + cursor_header_numbers.upsert(hash, &number)?; |
| 144 | + } |
| 145 | + } |
| 146 | + |
| 147 | + Ok(last_header_number) |
| 148 | +} |
0 commit comments