From c6fa510ac8d3a8e57311b9241a8a20f1197f913b Mon Sep 17 00:00:00 2001 From: ilitteri Date: Wed, 11 Feb 2026 14:08:43 -0300 Subject: [PATCH] Pipeline bytecode downloads and add background storage healing Two optimizations to the snap sync pipeline: 1. Concurrent bytecode downloads: Instead of downloading bytecodes sequentially after all healing completes, stream code hashes via an mpsc channel to a concurrent download task that runs alongside phases 3-7 (healing and storage). Bytecodes are content-addressed, so this is safe regardless of pivot changes. 2. Background storage healing: Split healing into state (must complete fully) and storage (can proceed at 99% threshold). Initial storage healing runs with a 5-minute time limit, then continues without limit until 99% of accounts are healed. After snap sync finalizes, remaining <1% of accounts heal in a background task while block execution starts. --- crates/networking/p2p/sync.rs | 3 + crates/networking/p2p/sync/code_collector.rs | 33 +- crates/networking/p2p/sync/healing/storage.rs | 33 +- crates/networking/p2p/sync/snap_sync.rs | 281 +++++++++++++----- 4 files changed, 269 insertions(+), 81 deletions(-) diff --git a/crates/networking/p2p/sync.rs b/crates/networking/p2p/sync.rs index 4ad64c9ac0d..767e63a7652 100644 --- a/crates/networking/p2p/sync.rs +++ b/crates/networking/p2p/sync.rs @@ -233,6 +233,8 @@ pub enum SyncError { RocksDBError(String), #[error("Bytecode file error")] BytecodeFileError, + #[error("Concurrent bytecode download task panicked: {0}")] + BytecodeTaskPanicked(String), #[error("Error in Peer Table: {0}")] PeerTableError(#[from] PeerTableError), #[error("Missing fullsync batch")] @@ -261,6 +263,7 @@ impl SyncError { | SyncError::StorageTempDBDirNotFound(_) | SyncError::RocksDBError(_) | SyncError::BytecodeFileError + | SyncError::BytecodeTaskPanicked(_) | SyncError::NoLatestCanonical | SyncError::PeerTableError(_) | SyncError::MissingFullsyncBatch diff --git a/crates/networking/p2p/sync/code_collector.rs b/crates/networking/p2p/sync/code_collector.rs index ced34ff88b1..20e3e077af4 100644 --- a/crates/networking/p2p/sync/code_collector.rs +++ b/crates/networking/p2p/sync/code_collector.rs @@ -6,6 +6,7 @@ use ethrex_common::H256; use ethrex_rlp::encode::RLPEncode; use std::collections::HashSet; use std::path::PathBuf; +use tokio::sync::mpsc; use tokio::task::JoinSet; use tracing::error; @@ -19,6 +20,8 @@ pub struct CodeHashCollector { file_index: u64, // JoinSet to manage async disk writes disk_tasks: JoinSet>, + // Optional channel to stream code hashes for concurrent bytecode downloading + code_hash_sender: Option>>, } impl CodeHashCollector { @@ -29,19 +32,38 @@ impl CodeHashCollector { snapshots_dir, file_index: 0, disk_tasks: JoinSet::new(), + code_hash_sender: None, } } - /// Adds a code hash to the buffer + /// Creates the channel and returns the receiver for concurrent bytecode downloading. + /// Must be called before any code hashes are added. + pub fn take_receiver(&mut self) -> mpsc::UnboundedReceiver> { + let (tx, rx) = mpsc::unbounded_channel(); + self.code_hash_sender = Some(tx); + rx + } + + /// Adds a code hash to the buffer and sends it through the channel if present pub fn add(&mut self, hash: H256) { - self.buffer.insert(hash); + if self.buffer.insert(hash) && let Some(sender) = &self.code_hash_sender { + // Ignore send errors — the receiver may have been dropped if + // the bytecode task finished early or errored out + let _ = sender.send(vec![hash]); + } } // The optimization for rocksdb database doesn't use this method #[cfg(not(feature = "rocksdb"))] /// Extends the buffer with a list of code hashes pub fn extend(&mut self, hashes: impl IntoIterator) { - self.buffer.extend(hashes); + let new_hashes: Vec = hashes + .into_iter() + .filter(|h| self.buffer.insert(*h)) + .collect(); + if !new_hashes.is_empty() && let Some(sender) = &self.code_hash_sender { + let _ = sender.send(new_hashes); + } } /// Flushes the buffer to a file if the buffer is larger than [`CODE_HASH_WRITE_BUFFER_SIZE`] @@ -55,7 +77,7 @@ impl CodeHashCollector { Ok(()) } - /// Finishes the code collector and returns the final index of file + /// Finishes the code collector: flushes remaining hashes and closes the channel pub async fn finish(mut self) -> Result<(), SyncError> { // Final flush if needed if !self.buffer.is_empty() { @@ -63,6 +85,9 @@ impl CodeHashCollector { self.flush_buffer(buffer); } + // Drop the sender to close the channel, signaling the bytecode task to drain + self.code_hash_sender.take(); + // Wait for all pending writes self.disk_tasks .join_all() diff --git a/crates/networking/p2p/sync/healing/storage.rs b/crates/networking/p2p/sync/healing/storage.rs index 29146acd4bb..407d0b0c479 100644 --- a/crates/networking/p2p/sync/healing/storage.rs +++ b/crates/networking/p2p/sync/healing/storage.rs @@ -116,6 +116,10 @@ pub struct NodeRequest { /// - When a node is downloaded: /// - if it has no missing children, we store it in the db /// - if the node has missing children, we store it in our healing_queue, which is preserved between calls +/// +/// When `time_limit` is `Some(duration)`, healing will stop early after the duration elapses +/// and return `Ok(false)`, allowing the caller to proceed with partial healing. +/// `global_roots_healed` accumulates the number of account storage roots fully healed across calls. pub async fn heal_storage_trie( state_root: H256, storage_accounts: &AccountStorageRoots, @@ -124,13 +128,17 @@ pub async fn heal_storage_trie( healing_queue: StorageHealingQueue, staleness_timestamp: u64, global_leafs_healed: &mut u64, + time_limit: Option, + global_roots_healed: &mut u64, ) -> Result { METRICS.current_step.set(CurrentStepValue::HealingStorage); let download_queue = get_initial_downloads(&store, state_root, storage_accounts); debug!( initial_accounts_count = download_queue.len(), + ?time_limit, "Started Storage Healing", ); + let healing_start_time = Instant::now(); let mut state = StorageHealer { last_update: Instant::now(), download_queue, @@ -203,8 +211,14 @@ pub async fn heal_storage_trie( let is_done = state.requests.is_empty() && state.download_queue.is_empty(); let is_stale = current_unix_time() > state.staleness_timestamp; - - if nodes_to_write.values().map(Vec::len).sum::() > 100_000 || is_done || is_stale { + let time_limit_reached = + time_limit.is_some_and(|limit| healing_start_time.elapsed() >= limit); + + if nodes_to_write.values().map(Vec::len).sum::() > 100_000 + || is_done + || is_stale + || time_limit_reached + { let to_write: Vec<_> = nodes_to_write.drain().collect(); let store = state.store.clone(); // NOTE: we keep only a single task in the background to avoid out of order deletes @@ -230,13 +244,24 @@ pub async fn heal_storage_trie( } if is_done { + *global_roots_healed += state.roots_healed as u64; db_joinset.join_all().await; return Ok(true); } - if is_stale { + if is_stale || time_limit_reached { + *global_roots_healed += state.roots_healed as u64; db_joinset.join_all().await; - state.healing_queue = HashMap::new(); + if is_stale { + state.healing_queue = HashMap::new(); + } + if time_limit_reached { + debug!( + roots_healed = state.roots_healed, + elapsed_ms = healing_start_time.elapsed().as_millis() as u64, + "Storage healing stopped: time limit reached", + ); + } return Ok(false); } diff --git a/crates/networking/p2p/sync/snap_sync.rs b/crates/networking/p2p/sync/snap_sync.rs index 19946596211..26fa19edf2b 100644 --- a/crates/networking/p2p/sync/snap_sync.rs +++ b/crates/networking/p2p/sync/snap_sync.rs @@ -22,6 +22,7 @@ use ethrex_storage::Store; #[cfg(feature = "rocksdb")] use ethrex_trie::Trie; use rayon::iter::{ParallelBridge, ParallelIterator}; +use tokio::sync::mpsc; use tracing::{debug, error, info, warn}; use crate::metrics::{CurrentStepValue, METRICS}; @@ -286,6 +287,14 @@ pub async fn snap_sync( let mut code_hash_collector: CodeHashCollector = CodeHashCollector::new(code_hashes_snapshot_dir.clone()); + // Set up channel to stream code hashes to a concurrent bytecode download task + let bytecode_rx = code_hash_collector.take_receiver(); + let bytecode_handle = tokio::spawn(download_bytecodes_concurrent( + peers.clone(), + store.clone(), + bytecode_rx, + )); + let mut storage_accounts = AccountStorageRoots::default(); if !std::env::var("SKIP_START_SNAP_SYNC").is_ok_and(|var| !var.is_empty()) { // We start by downloading all of the leafs of the trie of accounts @@ -441,9 +450,18 @@ pub async fn snap_sync( info!("Starting Healing Process"); let mut global_state_leafs_healed: u64 = 0; let mut global_storage_leafs_healed: u64 = 0; - let mut healing_done = false; - while !healing_done { - // This if is an edge case for the skip snap sync scenario + let mut global_roots_healed: u64 = 0; + let total_accounts_to_heal = storage_accounts.healed_accounts.len() as u64; + // Require >= 99% of storage account roots healed before proceeding to finalization. + // The remaining <1% will be healed in a background task while block execution starts. + let healing_threshold = (total_accounts_to_heal as f64 * 0.99).ceil() as u64; + info!( + "Storage healing target: {healing_threshold}/{total_accounts_to_heal} accounts (99% threshold)" + ); + + // Phase 1: Full state healing (must complete before storage healing) + let mut state_healing_done = false; + while !state_healing_done { if block_is_stale(&pivot_header) { pivot_header = update_pivot( pivot_header.number, @@ -453,7 +471,7 @@ pub async fn snap_sync( ) .await?; } - healing_done = heal_state_trie_wrap( + state_healing_done = heal_state_trie_wrap( pivot_header.state_root, store.clone(), peers, @@ -463,10 +481,65 @@ pub async fn snap_sync( &mut code_hash_collector, ) .await?; - if !healing_done { + } + info!("State healing complete, starting storage healing"); + + // Phase 2: Storage healing until threshold is met + // First pass: run with a 5-minute time limit to catch the bulk + let mut storage_fully_done = false; + if !storage_fully_done && global_roots_healed < healing_threshold { + if block_is_stale(&pivot_header) { + pivot_header = update_pivot( + pivot_header.number, + pivot_header.timestamp, + peers, + block_sync_state, + ) + .await?; + } + storage_fully_done = heal_storage_trie( + pivot_header.state_root, + &storage_accounts, + peers, + store.clone(), + HashMap::new(), + calculate_staleness_timestamp(pivot_header.timestamp), + &mut global_storage_leafs_healed, + Some(Duration::from_secs(300)), + &mut global_roots_healed, + ) + .await?; + info!( + "Initial storage healing pass: {global_roots_healed}/{total_accounts_to_heal} roots healed" + ); + } + + // Continue healing without time limit until threshold is reached or fully done + while !storage_fully_done && global_roots_healed < healing_threshold { + if block_is_stale(&pivot_header) { + pivot_header = update_pivot( + pivot_header.number, + pivot_header.timestamp, + peers, + block_sync_state, + ) + .await?; + } + // Re-do state healing if pivot changed + let state_ok = heal_state_trie_wrap( + pivot_header.state_root, + store.clone(), + peers, + calculate_staleness_timestamp(pivot_header.timestamp), + &mut global_state_leafs_healed, + &mut storage_accounts, + &mut code_hash_collector, + ) + .await?; + if !state_ok { continue; } - healing_done = heal_storage_trie( + storage_fully_done = heal_storage_trie( pivot_header.state_root, &storage_accounts, peers, @@ -474,90 +547,46 @@ pub async fn snap_sync( HashMap::new(), calculate_staleness_timestamp(pivot_header.timestamp), &mut global_storage_leafs_healed, + None, + &mut global_roots_healed, ) .await?; + info!( + "Storage healing progress: {global_roots_healed}/{total_accounts_to_heal} roots healed" + ); } *METRICS.heal_end_time.lock().await = Some(SystemTime::now()); + info!( + "Storage healing sufficient: {global_roots_healed}/{total_accounts_to_heal} roots healed (fully done: {storage_fully_done})" + ); store.generate_flatkeyvalue()?; debug_assert!(validate_state_root(store.clone(), pivot_header.state_root).await); - debug_assert!(validate_storage_root(store.clone(), pivot_header.state_root).await); + // Only validate storage root if healing fully completed — background healing + // handles the remaining <1% of accounts after snap sync finalization + if storage_fully_done { + debug_assert!(validate_storage_root(store.clone(), pivot_header.state_root).await); + } info!("Finished healing"); - // Finish code hash collection + // Finish code hash collection — this drops the channel sender, signaling + // the concurrent bytecode task to drain remaining hashes and complete code_hash_collector.finish().await?; - *METRICS.bytecode_download_start_time.lock().await = Some(SystemTime::now()); + info!("Waiting for concurrent bytecode download to complete"); + bytecode_handle + .await + .map_err(|e| SyncError::BytecodeTaskPanicked(e.to_string()))??; + // Clean up code hash snapshot files (written by CodeHashCollector as backup) let code_hashes_dir = get_code_hashes_snapshots_dir(datadir); - let mut seen_code_hashes = HashSet::new(); - let mut code_hashes_to_download = Vec::new(); - - info!("Starting download code hashes from peers"); - for entry in std::fs::read_dir(&code_hashes_dir) - .map_err(|_| SyncError::CodeHashesSnapshotsDirNotFound)? - { - let entry = entry.map_err(|_| SyncError::CorruptPath)?; - let snapshot_contents = std::fs::read(entry.path()) - .map_err(|err| SyncError::SnapshotReadError(entry.path(), err))?; - let code_hashes: Vec = RLPDecode::decode(&snapshot_contents) - .map_err(|_| SyncError::CodeHashesSnapshotDecodeError(entry.path()))?; - - for hash in code_hashes { - // If we haven't seen the code hash yet, add it to the list of hashes to download - if seen_code_hashes.insert(hash) { - code_hashes_to_download.push(hash); - - if code_hashes_to_download.len() >= BYTECODE_CHUNK_SIZE { - info!( - "Starting bytecode download of {} hashes", - code_hashes_to_download.len() - ); - let bytecodes = request_bytecodes(peers, &code_hashes_to_download) - .await? - .ok_or(SyncError::BytecodesNotFound)?; - - store - .write_account_code_batch_no_wal( - code_hashes_to_download - .drain(..) - .zip(bytecodes) - // SAFETY: hash already checked by the download worker - .map(|(hash, code)| { - (hash, Code::from_bytecode_unchecked(code, hash)) - }) - .collect(), - ) - .await?; - } - } - } - } - - // Download remaining bytecodes if any - if !code_hashes_to_download.is_empty() { - let bytecodes = request_bytecodes(peers, &code_hashes_to_download) - .await? - .ok_or(SyncError::BytecodesNotFound)?; - store - .write_account_code_batch_no_wal( - code_hashes_to_download - .drain(..) - .zip(bytecodes) - // SAFETY: hash already checked by the download worker - .map(|(hash, code)| (hash, Code::from_bytecode_unchecked(code, hash))) - .collect(), - ) - .await?; + if code_hashes_dir.exists() { + std::fs::remove_dir_all(code_hashes_dir) + .map_err(|_| SyncError::CodeHashesSnapshotsDirNotFound)?; } - std::fs::remove_dir_all(code_hashes_dir) - .map_err(|_| SyncError::CodeHashesSnapshotsDirNotFound)?; - - *METRICS.bytecode_download_end_time.lock().await = Some(SystemTime::now()); - debug_assert!(validate_bytecodes(store.clone(), pivot_header.state_root)); store_block_bodies(vec![pivot_header.clone()], peers.clone(), store.clone()).await?; @@ -586,6 +615,112 @@ pub async fn snap_sync( None, ) .await?; + + // Spawn background storage healing for remaining un-healed accounts (<1%) + if !storage_fully_done { + let remaining = total_accounts_to_heal.saturating_sub(global_roots_healed); + info!( + "Spawning background storage healing for ~{remaining} remaining accounts" + ); + let bg_state_root = pivot_header.state_root; + // Only clone the healed_accounts set — accounts_with_storage_root is not needed + // by heal_storage_trie (get_initial_downloads only reads healed_accounts) + let bg_storage_accounts = AccountStorageRoots { + accounts_with_storage_root: std::collections::BTreeMap::new(), + healed_accounts: storage_accounts.healed_accounts.clone(), + }; + let mut bg_peers = peers.clone(); + let bg_store = store.clone(); + tokio::spawn(async move { + let mut bg_leafs_healed: u64 = 0; + let mut bg_roots_healed: u64 = 0; + match heal_storage_trie( + bg_state_root, + &bg_storage_accounts, + &mut bg_peers, + bg_store, + HashMap::new(), + u64::MAX, // no staleness limit for background healing + &mut bg_leafs_healed, + None, // no time limit + &mut bg_roots_healed, + ) + .await + { + Ok(_) => info!( + "Background storage healing completed: {bg_roots_healed} roots, {bg_leafs_healed} leaves healed" + ), + Err(e) => warn!("Background storage healing failed: {e}"), + } + }); + } + + Ok(()) +} + +/// Downloads bytecodes concurrently as code hashes arrive through the channel. +/// Runs as a spawned task alongside phases 2-7 (insert accounts, healing, storage). +/// Completes when the channel sender is dropped (via `CodeHashCollector::finish()`). +async fn download_bytecodes_concurrent( + mut peers: PeerHandler, + store: Store, + mut rx: mpsc::UnboundedReceiver>, +) -> Result<(), SyncError> { + let mut seen = HashSet::new(); + let mut batch = Vec::new(); + let mut total_downloaded: u64 = 0; + + *METRICS.bytecode_download_start_time.lock().await = Some(SystemTime::now()); + + while let Some(hashes) = rx.recv().await { + for hash in hashes { + if hash != *EMPTY_KECCACK_HASH && seen.insert(hash) { + batch.push(hash); + if batch.len() >= BYTECODE_CHUNK_SIZE { + download_and_store_bytecodes(&mut peers, &store, &mut batch, &mut total_downloaded).await?; + } + } + } + } + + // Drain remaining batch after channel closed + if !batch.is_empty() { + download_and_store_bytecodes(&mut peers, &store, &mut batch, &mut total_downloaded).await?; + } + + *METRICS.bytecode_download_end_time.lock().await = Some(SystemTime::now()); + info!("Concurrent bytecode download complete: {total_downloaded} bytecodes downloaded"); + Ok(()) +} + +/// Helper to download a batch of bytecodes and write them to the store +async fn download_and_store_bytecodes( + peers: &mut PeerHandler, + store: &Store, + batch: &mut Vec, + total_downloaded: &mut u64, +) -> Result<(), SyncError> { + info!( + "Downloading bytecode batch of {} hashes (total so far: {})", + batch.len(), + *total_downloaded + ); + let bytecodes = request_bytecodes(peers, batch) + .await? + .ok_or(SyncError::BytecodesNotFound)?; + + *total_downloaded += bytecodes.len() as u64; + + store + .write_account_code_batch_no_wal( + batch + .drain(..) + .zip(bytecodes) + // SAFETY: hash already checked by the download worker + .map(|(hash, code)| (hash, Code::from_bytecode_unchecked(code, hash))) + .collect(), + ) + .await?; Ok(()) }