ethrex_p2p/snap/

client.rs

//! Snap sync client - functions for requesting snap protocol data from peers
//!
//! This module contains all the client-side snap protocol request functions.

use crate::rlpx::message::Message as RLPxMessage;
use crate::{
    metrics::{CurrentStepValue, METRICS},
    peer_handler::PeerHandler,
    peer_table::{PeerTableServerProtocol as _, RequestPermit},
    rlpx::{
        connection::server::PeerConnection,
        error::PeerConnectionError,
        p2p::SUPPORTED_SNAP_CAPABILITIES,
        snap::{
            AccountRange, AccountRangeUnit, ByteCodes, GetAccountRange, GetByteCodes,
            GetStorageRanges, GetTrieNodes, StorageRanges, TrieNodes,
        },
    },
    snap::{async_fs, constants::*, encodable_to_proof, error::SnapError},
    sync::{AccountStorageRoots, SnapBlockSyncState, block_is_stale, update_pivot},
    utils::{
        AccountsWithStorage, dump_accounts_to_file, dump_storages_to_file,
        get_account_state_snapshot_file, get_account_storages_snapshot_file,
    },
};
use bytes::Bytes;
use ethrex_common::{
    BigEndianHash, H256, U256,
    types::{AccountState, BlockHeader},
};
use ethrex_crypto::NativeCrypto;
use ethrex_rlp::{decode::RLPDecode, encode::RLPEncode};
use ethrex_storage::Store;
use ethrex_trie::Nibbles;
use ethrex_trie::{Node, verify_range};
use std::{
    collections::{BTreeMap, HashMap, VecDeque},
    path::Path,
    sync::atomic::Ordering,
    time::{Duration, SystemTime},
};
use tracing::{debug, error, trace, warn};

// Re-export DumpError from error module
pub use super::error::DumpError;

/// Metadata for requesting trie nodes
#[derive(Debug, Clone)]
pub struct RequestMetadata {
    pub hash: H256,
    pub path: Nibbles,
    /// What node is the parent of this node
    pub parent_path: Nibbles,
}

/// Error type for storage trie node requests (includes request ID for tracking)
#[derive(Debug, thiserror::Error)]
#[error("Storage trie node request {request_id} failed: {source}")]
pub struct RequestStorageTrieNodesError {
    pub request_id: u64,
    #[source]
    pub source: SnapError,
}

struct StorageTaskResult {
    start_index: usize,
    account_storages: Vec<Vec<(H256, U256)>>,
    peer_id: H256,
    remaining_start: usize,
    remaining_end: usize,
    remaining_hash_range: (H256, Option<H256>),
    // The start_hash of the original task. Distinct from remaining_hash_range.0,
    // which is the worker's advancing pointer (zero on full completion). Needed
    // by the response handler to match the completed interval unambiguously.
    task_start_hash: H256,
}

#[derive(Debug)]
struct StorageTask {
    start_index: usize,
    end_index: usize,
    start_hash: H256,
    // end_hash is None if the task is for the first big storage request
    end_hash: Option<H256>,
}

/// Removes the completed interval `(start_hash, end_hash)` from whichever
/// account in the group at `accounts_by_root_hash[start_index]` currently
/// holds the interval list, and when the list empties marks every account in
/// the group as done and healed.
///
/// Returns `true` if an interval was found and removed, `false` if no account
/// in the group has any live intervals (a sibling task already finalized the
/// account earlier in this call).
///
/// Within a group sharing the same storage root, the split path stores
/// intervals under one canonical account, so only that account's entry holds
/// the live list. We scan the group rather than relying on `accounts.first()`
/// because the canonical account can shift across calls if the set of tracked
/// accounts changes between iterations of `accounts_with_storage_root`.
fn clear_completed_interval(
    account_storage_roots: &mut AccountStorageRoots,
    accounts_by_root_hash: &[(H256, Vec<H256>)],
    accounts_done: &mut HashMap<H256, Vec<(H256, H256)>>,
    start_index: usize,
    interval: (H256, H256),
) -> Result<bool, SnapError> {
    let accounts = &accounts_by_root_hash[start_index].1;
    let acc_hash = accounts.iter().copied().find(|account| {
        account_storage_roots
            .accounts_with_storage_root
            .get(account)
            .is_some_and(|(_, ivs)| !ivs.is_empty())
    });
    let Some(acc_hash) = acc_hash else {
        return Ok(false);
    };
    let (_, old_intervals) = account_storage_roots
        .accounts_with_storage_root
        .get_mut(&acc_hash)
        .ok_or_else(|| {
            SnapError::InternalError(
                "Tried to get the old download intervals for an account but did not find them"
                    .to_owned(),
            )
        })?;
    let pos = old_intervals
        .iter()
        .position(|iv| *iv == interval)
        .ok_or_else(|| {
            SnapError::InternalError(
                "Could not find an old interval that we were tracking".to_owned(),
            )
        })?;
    old_intervals.remove(pos);
    if old_intervals.is_empty() {
        for account in accounts {
            accounts_done.insert(*account, vec![]);
            account_storage_roots.healed_accounts.insert(*account);
        }
    }
    Ok(true)
}

/// Requests an account range from any suitable peer given the state trie's root and the starting hash and the limit hash.
/// Will also return a boolean indicating if there is more state to be fetched towards the right of the trie
/// (Note that the boolean will be true even if the remaining state is ouside the boundary set by the limit hash)
///
/// # Returns
///
/// The account range or `None` if:
///
/// - There are no available peers (the node just started up or was rejected by all other nodes)
/// - No peer returned a valid response in the given time and retry limits
pub async fn request_account_range(
    peers: &mut PeerHandler,
    start: H256,
    limit: H256,
    account_state_snapshots_dir: &Path,
    pivot_header: &mut BlockHeader,
    block_sync_state: &mut SnapBlockSyncState,
    diagnostics: &std::sync::Arc<tokio::sync::RwLock<crate::sync::SyncDiagnostics>>,
) -> Result<(), SnapError> {
    METRICS
        .current_step
        .set(CurrentStepValue::RequestingAccountRanges);
    // 1) split the range in chunks of same length
    let start_u256 = U256::from_big_endian(&start.0);
    let limit_u256 = U256::from_big_endian(&limit.0);

    let range = limit_u256 - start_u256;
    // Bounded by ACCOUNT_RANGE_CHUNK_COUNT (800), always fits in usize.
    let chunk_count =
        usize::try_from(U256::from(ACCOUNT_RANGE_CHUNK_COUNT).min(range.max(U256::one())))
            .expect("chunk_count bounded by ACCOUNT_RANGE_CHUNK_COUNT");
    let chunk_size = range / chunk_count;

    // list of tasks to be executed
    let mut tasks_queue_not_started = VecDeque::<(H256, H256)>::new();
    for i in 0..(chunk_count as u64) {
        let chunk_start_u256 = chunk_size * i + start_u256;
        // We subtract one because ranges are inclusive
        let chunk_end_u256 = chunk_start_u256 + chunk_size - 1u64;
        let chunk_start = H256::from_uint(&(chunk_start_u256));
        let chunk_end = H256::from_uint(&(chunk_end_u256));
        tasks_queue_not_started.push_back((chunk_start, chunk_end));
    }
    // Modify the last chunk to include the limit
    let last_task = tasks_queue_not_started
        .back_mut()
        .ok_or(SnapError::NoTasks)?;
    last_task.1 = limit;

    // 2) request the chunks from peers

    let mut downloaded_count = 0_u64;
    let mut all_account_hashes = Vec::new();
    let mut all_accounts_state = Vec::new();

    // channel to send the tasks to the peers
    let (task_sender, mut task_receiver) =
        tokio::sync::mpsc::channel::<(Vec<AccountRangeUnit>, H256, Option<(H256, H256)>)>(1000);

    debug!("Starting to download account ranges from peers");

    *METRICS.account_tries_download_start_time.lock().await = Some(SystemTime::now());

    let mut completed_tasks = 0;
    let mut chunk_file = 0;
    let mut last_update: SystemTime = SystemTime::now();
    let mut write_set = tokio::task::JoinSet::new();

    let mut logged_no_free_peers_count = 0;

    loop {
        if all_accounts_state.len() * size_of::<AccountState>() >= RANGE_FILE_CHUNK_SIZE {
            let current_account_hashes = std::mem::take(&mut all_account_hashes);
            let current_account_states = std::mem::take(&mut all_accounts_state);

            let account_state_chunk = current_account_hashes
                .into_iter()
                .zip(current_account_states)
                .collect::<Vec<(H256, AccountState)>>();

            async_fs::ensure_dir_exists(account_state_snapshots_dir).await?;

            let account_state_snapshots_dir_cloned = account_state_snapshots_dir.to_path_buf();
            write_set.spawn(async move {
                let path = get_account_state_snapshot_file(
                    &account_state_snapshots_dir_cloned,
                    chunk_file,
                );
                // TODO: check the error type and handle it properly
                dump_accounts_to_file(&path, account_state_chunk)
            });

            chunk_file += 1;
        }

        if last_update
            .elapsed()
            .expect("Time shouldn't be in the past")
            >= Duration::from_secs(1)
        {
            METRICS
                .downloaded_account_tries
                .store(downloaded_count, Ordering::Relaxed);
            last_update = SystemTime::now();
        }

        if let Ok((accounts, peer_id, chunk_start_end)) = task_receiver.try_recv() {
            if let Some((chunk_start, chunk_end)) = chunk_start_end {
                if chunk_start <= chunk_end {
                    tasks_queue_not_started.push_back((chunk_start, chunk_end));
                } else {
                    completed_tasks += 1;
                }
            }
            if chunk_start_end.is_none() {
                completed_tasks += 1;
            }
            if accounts.is_empty() {
                peers.peer_table.record_failure(peer_id)?;
                continue;
            }
            peers.peer_table.record_success(peer_id)?;

            downloaded_count += accounts.len() as u64;

            debug!(
                "Downloaded {} accounts from peer {} (current count: {downloaded_count})",
                accounts.len(),
                peer_id
            );
            all_account_hashes.extend(accounts.iter().map(|unit| unit.hash));
            all_accounts_state.extend(accounts.iter().map(|unit| unit.account));
        }

        let Some((peer_id, connection, permit)) = peers
            .peer_table
            .get_best_peer(SUPPORTED_SNAP_CAPABILITIES.to_vec())
            .await
            .inspect_err(|err| warn!(%err, "Error requesting a peer for account range"))
            .unwrap_or(None)
        else {
            // Log ~ once every 10 seconds
            if logged_no_free_peers_count == 0 {
                trace!("We are missing peers in request_account_range");
                logged_no_free_peers_count = 1000;
            }
            logged_no_free_peers_count -= 1;
            // Sleep a bit to avoid busy polling
            tokio::time::sleep(Duration::from_millis(10)).await;
            continue;
        };

        let Some((chunk_start, chunk_end)) = tasks_queue_not_started.pop_front() else {
            if completed_tasks >= chunk_count {
                debug!("All account ranges downloaded successfully");
                break;
            }
            continue;
        };

        let tx = task_sender.clone();

        if block_is_stale(pivot_header) {
            debug!("Pivot became stale during account range download, updating pivot");
            *pivot_header = update_pivot(
                pivot_header.number,
                pivot_header.timestamp,
                peers,
                block_sync_state,
                diagnostics,
            )
            .await
            .expect("Should be able to update pivot")
        }

        tokio::spawn(request_account_range_worker(
            peer_id,
            connection,
            chunk_start,
            chunk_end,
            pivot_header.state_root,
            tx,
            permit,
        ));
    }

    write_set
        .join_all()
        .await
        .into_iter()
        .collect::<Result<Vec<()>, DumpError>>()
        .map_err(SnapError::from)?;

    // TODO: This is repeated code, consider refactoring
    {
        let current_account_hashes = std::mem::take(&mut all_account_hashes);
        let current_account_states = std::mem::take(&mut all_accounts_state);

        let account_state_chunk = current_account_hashes
            .into_iter()
            .zip(current_account_states)
            .collect::<Vec<(H256, AccountState)>>();

        async_fs::ensure_dir_exists(account_state_snapshots_dir).await?;

        let path = get_account_state_snapshot_file(account_state_snapshots_dir, chunk_file);
        dump_accounts_to_file(&path, account_state_chunk)
            .inspect_err(|err| error!("Failed to dump remaining accounts to disk: {}", err.error))
            .map_err(|_| {
                SnapError::SnapshotDir(format!(
                    "Failed to write state snapshot chunk {}",
                    chunk_file
                ))
            })?;
    }

    METRICS
        .downloaded_account_tries
        .store(downloaded_count, Ordering::Relaxed);
    *METRICS.account_tries_download_end_time.lock().await = Some(SystemTime::now());

    Ok(())
}

/// Requests bytecodes for the given code hashes
/// Returns the bytecodes or None if:
/// - There are no available peers (the node just started up or was rejected by all other nodes)
/// - No peer returned a valid response in the given time and retry limits
pub async fn request_bytecodes(
    peers: &mut PeerHandler,
    all_bytecode_hashes: &[H256],
) -> Result<Option<Vec<Bytes>>, SnapError> {
    METRICS
        .current_step
        .set(CurrentStepValue::RequestingBytecodes);
    if all_bytecode_hashes.is_empty() {
        return Ok(Some(Vec::new()));
    }
    const MAX_BYTECODES_REQUEST_SIZE: usize = 100;
    // 1) split the range in chunks of same length
    let chunk_count = 800;
    let chunk_count = chunk_count.min(all_bytecode_hashes.len());
    let chunk_size = all_bytecode_hashes.len() / chunk_count;

    // list of tasks to be executed
    // Types are (start_index, end_index, starting_hash)
    // NOTE: end_index is NOT inclusive
    let mut tasks_queue_not_started = VecDeque::<(usize, usize)>::new();
    for i in 0..chunk_count {
        let chunk_start = chunk_size * i;
        let chunk_end = chunk_start + chunk_size;
        tasks_queue_not_started.push_back((chunk_start, chunk_end));
    }
    // Modify the last chunk to include the limit
    let last_task = tasks_queue_not_started
        .back_mut()
        .ok_or(SnapError::NoTasks)?;
    last_task.1 = all_bytecode_hashes.len();

    // 2) request the chunks from peers
    let mut downloaded_count = 0_u64;
    let mut all_bytecodes = vec![Bytes::new(); all_bytecode_hashes.len()];

    // channel to send the tasks to the peers
    struct TaskResult {
        start_index: usize,
        bytecodes: Vec<Bytes>,
        peer_id: H256,
        remaining_start: usize,
        remaining_end: usize,
    }
    let (task_sender, mut task_receiver) = tokio::sync::mpsc::channel::<TaskResult>(1000);

    debug!("Starting to download bytecodes from peers");

    METRICS
        .bytecodes_to_download
        .fetch_add(all_bytecode_hashes.len() as u64, Ordering::Relaxed);

    let mut completed_tasks = 0;

    let mut logged_no_free_peers_count = 0;

    loop {
        if let Ok(result) = task_receiver.try_recv() {
            let TaskResult {
                start_index,
                bytecodes,
                peer_id,
                remaining_start,
                remaining_end,
            } = result;

            debug!(
                "Downloaded {} bytecodes from peer {peer_id} (current count: {downloaded_count})",
                bytecodes.len(),
            );

            if remaining_start < remaining_end {
                tasks_queue_not_started.push_back((remaining_start, remaining_end));
            } else {
                completed_tasks += 1;
            }
            if bytecodes.is_empty() {
                peers.peer_table.record_failure(peer_id)?;
                continue;
            }

            downloaded_count += bytecodes.len() as u64;

            peers.peer_table.record_success(peer_id)?;
            for (i, bytecode) in bytecodes.into_iter().enumerate() {
                all_bytecodes[start_index + i] = bytecode;
            }
        }

        let Some((peer_id, mut connection, permit)) = peers
            .peer_table
            .get_best_peer(SUPPORTED_SNAP_CAPABILITIES.to_vec())
            .await
            .inspect_err(|err| warn!(%err, "Error requesting a peer for bytecodes"))
            .unwrap_or(None)
        else {
            // Log ~ once every 10 seconds
            if logged_no_free_peers_count == 0 {
                trace!("We are missing peers in request_bytecodes");
                logged_no_free_peers_count = 1000;
            }
            logged_no_free_peers_count -= 1;
            // Sleep a bit to avoid busy polling
            tokio::time::sleep(Duration::from_millis(10)).await;
            continue;
        };

        let Some((chunk_start, chunk_end)) = tasks_queue_not_started.pop_front() else {
            if completed_tasks >= chunk_count {
                debug!("All bytecodes downloaded successfully");
                break;
            }
            continue;
        };

        let tx = task_sender.clone();

        let hashes_to_request: Vec<_> = all_bytecode_hashes
            .iter()
            .skip(chunk_start)
            .take((chunk_end - chunk_start).min(MAX_BYTECODES_REQUEST_SIZE))
            .copied()
            .collect();

        tokio::spawn(async move {
            debug!(
                "Requesting bytecode from peer {peer_id}, chunk: {chunk_start:?} - {chunk_end:?}"
            );
            let request_id = rand::random();
            let request = RLPxMessage::GetByteCodes(GetByteCodes {
                id: request_id,
                hashes: hashes_to_request.clone(),
                bytes: MAX_RESPONSE_BYTES,
            });

            let response = connection
                .outgoing_request(request, PEER_REPLY_TIMEOUT)
                .await;
            drop(permit);

            let (bytecodes, remaining_start) = match response {
                Ok(RLPxMessage::ByteCodes(ByteCodes { id: _, codes })) if !codes.is_empty() => {
                    let validated_codes: Vec<Bytes> = codes
                        .into_iter()
                        .zip(hashes_to_request)
                        .take_while(|(b, hash)| ethrex_common::utils::keccak(b) == *hash)
                        .map(|(b, _hash)| b)
                        .collect();
                    let new_remaining_start = chunk_start + validated_codes.len();
                    (validated_codes, new_remaining_start)
                }
                Ok(RLPxMessage::ByteCodes(_)) => {
                    // Empty response; retry the full chunk.
                    (Vec::new(), chunk_start)
                }
                _ => {
                    tracing::debug!("Failed to get bytecode");
                    (Vec::new(), chunk_start)
                }
            };

            let result = TaskResult {
                start_index: chunk_start,
                bytecodes,
                peer_id,
                remaining_start,
                remaining_end: chunk_end,
            };
            tx.send(result).await.ok();
        });
    }

    METRICS
        .downloaded_bytecodes
        .fetch_add(downloaded_count, Ordering::Relaxed);
    debug!(
        "Finished downloading bytecodes, total bytecodes: {}",
        all_bytecode_hashes.len()
    );

    Ok(Some(all_bytecodes))
}

/// Requests storage ranges for accounts given their hashed address and storage roots, and the root of their state trie
/// account_hashes & storage_roots must have the same length
/// storage_roots must not contain empty trie hashes, we will treat empty ranges as invalid responses
/// Returns true if the last account's storage was not completely fetched by the request
/// Returns the list of hashed storage keys and values for each account's storage or None if:
/// - There are no available peers (the node just started up or was rejected by all other nodes)
/// - No peer returned a valid response in the given time and retry limits
pub async fn request_storage_ranges(
    peers: &mut PeerHandler,
    account_storage_roots: &mut AccountStorageRoots,
    account_storages_snapshots_dir: &Path,
    mut chunk_index: u64,
    pivot_header: &mut BlockHeader,
    store: Store,
) -> Result<u64, SnapError> {
    METRICS
        .current_step
        .set(CurrentStepValue::RequestingStorageRanges);
    debug!("Starting request_storage_ranges function");
    // 1) split the range in chunks of same length
    let mut accounts_by_root_hash: BTreeMap<_, Vec<_>> = BTreeMap::new();
    for (account, (maybe_root_hash, _)) in &account_storage_roots.accounts_with_storage_root {
        match maybe_root_hash {
            Some(root) => {
                accounts_by_root_hash
                    .entry(*root)
                    .or_default()
                    .push(*account);
            }
            None => {
                let root = store
                    .get_account_state_by_acc_hash(pivot_header.hash(), *account)?
                    .ok_or_else(|| {
                        SnapError::InternalError(
                            "Could not find account that should have been downloaded or healed"
                                .to_string(),
                        )
                    })?
                    .storage_root;
                accounts_by_root_hash
                    .entry(root)
                    .or_default()
                    .push(*account);
            }
        }
    }
    // Invariant: within a group sharing the same storage root, the split path
    // stores intervals under one canonical account, so at most one account's
    // entry in `accounts_with_storage_root` holds a non-empty interval list.
    // Scheduling and completion code below scan the group to find that account
    // rather than relying on `accounts.first()`, because iteration order of
    // `accounts_with_storage_root` can shift between calls when the tracked
    // set changes.
    let mut accounts_by_root_hash = Vec::from_iter(accounts_by_root_hash);
    // TODO: Turn this into a stable sort for binary search.
    accounts_by_root_hash.sort_unstable_by_key(|(_, accounts)| !accounts.len());
    let chunk_size = STORAGE_BATCH_SIZE;

    // Partition into bulk-path tasks (fresh accounts with empty intervals) and
    // per-interval tasks (big accounts marked in a prior call). The previous
    // implementation queued every account from `start_hash: zero` and relied
    // on the response handler's bulk-task big-account split path to re-queue
    // per-interval tasks each call. That fails when peers cover a big account
    // fully without hitting their response limit on it: the split path doesn't
    // fire, no per-interval tasks get queued, intervals never drain, the
    // account is stuck pending forever even after its data is on disk.
    let mut tasks_queue_not_started = VecDeque::<StorageTask>::new();
    let mut bulk_chunk_start: Option<usize> = None;
    for (i, (_, accounts)) in accounts_by_root_hash.iter().enumerate() {
        let intervals = accounts.iter().find_map(|acc| {
            account_storage_roots
                .accounts_with_storage_root
                .get(acc)
                .and_then(|(_, ivs)| (!ivs.is_empty()).then_some(ivs))
        });
        if let Some(intervals) = intervals {
            if let Some(start) = bulk_chunk_start.take() {
                tasks_queue_not_started.push_back(StorageTask {
                    start_index: start,
                    end_index: i,
                    start_hash: H256::zero(),
                    end_hash: None,
                });
            }
            for &(start_hash, end_hash) in intervals.iter() {
                tasks_queue_not_started.push_back(StorageTask {
                    start_index: i,
                    end_index: i + 1,
                    start_hash,
                    end_hash: Some(end_hash),
                });
            }
        } else {
            let chunk_start = *bulk_chunk_start.get_or_insert(i);
            if i + 1 - chunk_start >= chunk_size {
                tasks_queue_not_started.push_back(StorageTask {
                    start_index: chunk_start,
                    end_index: i + 1,
                    start_hash: H256::zero(),
                    end_hash: None,
                });
                bulk_chunk_start = None;
            }
        }
    }
    if let Some(start) = bulk_chunk_start {
        tasks_queue_not_started.push_back(StorageTask {
            start_index: start,
            end_index: accounts_by_root_hash.len(),
            start_hash: H256::zero(),
            end_hash: None,
        });
    }

    // channel to send the tasks to the peers
    let (task_sender, mut task_receiver) = tokio::sync::mpsc::channel::<StorageTaskResult>(1000);

    // channel to send the result of dumping storages
    let mut disk_joinset: tokio::task::JoinSet<Result<(), DumpError>> = tokio::task::JoinSet::new();

    let mut task_count = tasks_queue_not_started.len();
    let mut completed_tasks = 0;

    // TODO: in a refactor, delete this replace with a structure that can handle removes
    let mut accounts_done: HashMap<H256, Vec<(H256, H256)>> = HashMap::new();
    // Maps storage root to vector of hashed addresses matching that root and
    // vector of hashed storage keys and storage values.
    let mut current_account_storages: BTreeMap<H256, AccountsWithStorage> = BTreeMap::new();

    let mut logged_no_free_peers_count = 0;

    debug!("Starting request_storage_ranges loop");
    loop {
        if current_account_storages
            .values()
            .map(|accounts| 32 * accounts.accounts.len() + 64 * accounts.storages.len())
            .sum::<usize>()
            > RANGE_FILE_CHUNK_SIZE
        {
            let current_account_storages = std::mem::take(&mut current_account_storages);
            let snapshot = current_account_storages.into_values().collect::<Vec<_>>();

            async_fs::ensure_dir_exists(account_storages_snapshots_dir).await?;

            let account_storages_snapshots_dir_cloned =
                account_storages_snapshots_dir.to_path_buf();
            if !disk_joinset.is_empty() {
                debug!("Writing to disk");
                disk_joinset
                    .join_next()
                    .await
                    .expect("Shouldn't be empty")
                    .expect("Shouldn't have a join error")
                    .inspect_err(|err| error!("Failed to dump storage snapshot to file: {err:?}"))
                    .map_err(SnapError::from)?;
            }
            disk_joinset.spawn(async move {
                let path = get_account_storages_snapshot_file(
                    &account_storages_snapshots_dir_cloned,
                    chunk_index,
                );
                dump_storages_to_file(&path, snapshot)
            });

            chunk_index += 1;
        }

        if let Ok(result) = task_receiver.try_recv() {
            let StorageTaskResult {
                start_index,
                mut account_storages,
                peer_id,
                remaining_start,
                remaining_end,
                remaining_hash_range: (hash_start, hash_end),
                task_start_hash,
            } = result;
            completed_tasks += 1;

            for (_, accounts) in accounts_by_root_hash[start_index..remaining_start].iter() {
                for account in accounts {
                    if !accounts_done.contains_key(account) {
                        let (_, old_intervals) = account_storage_roots
                                .accounts_with_storage_root
                                .get_mut(account)
                                .ok_or(SnapError::InternalError("Tried to get the old download intervals for an account but did not find them".to_owned()))?;

                        if old_intervals.is_empty() {
                            accounts_done.insert(*account, vec![]);
                        }
                    }
                }
            }

            if remaining_start < remaining_end {
                debug!("Failed to download entire chunk from peer {peer_id}");
                if hash_start.is_zero() {
                    // Task is common storage range request
                    let task = StorageTask {
                        start_index: remaining_start,
                        end_index: remaining_end,
                        start_hash: H256::zero(),
                        end_hash: None,
                    };
                    tasks_queue_not_started.push_back(task);
                    task_count += 1;
                } else if let Some(hash_end) = hash_end {
                    // Task was a big storage account result
                    if hash_start <= hash_end {
                        let task = StorageTask {
                            start_index: remaining_start,
                            end_index: remaining_end,
                            start_hash: hash_start,
                            end_hash: Some(hash_end),
                        };
                        tasks_queue_not_started.push_back(task);
                        task_count += 1;

                        let acc_hash = *accounts_by_root_hash[remaining_start]
                            .1
                            .first()
                            .ok_or(SnapError::InternalError("Empty accounts vector".to_owned()))?;
                        let (_, old_intervals) = account_storage_roots
                                .accounts_with_storage_root
                                .get_mut(&acc_hash).ok_or(SnapError::InternalError("Tried to get the old download intervals for an account but did not find them".to_owned()))?;
                        for (old_start, end) in old_intervals {
                            if *old_start == task_start_hash && *end == hash_end {
                                *old_start = hash_start;
                                break;
                            }
                        }
                        account_storage_roots
                            .healed_accounts
                            .extend(accounts_by_root_hash[start_index].1.iter().copied());
                    } else {
                        // Peer overran the original interval limit; the original
                        // task's interval is fully covered. Remaining work in
                        // this chunk still exists, so no sibling task has
                        // drained the account yet — a missing acc_hash here
                        // indicates corruption.
                        let found = clear_completed_interval(
                            account_storage_roots,
                            &accounts_by_root_hash,
                            &mut accounts_done,
                            remaining_start,
                            (task_start_hash, hash_end),
                        )?;
                        if !found {
                            panic!("Should have found the account hash");
                        }
                    }
                } else {
                    if remaining_start + 1 < remaining_end {
                        let task = StorageTask {
                            start_index: remaining_start + 1,
                            end_index: remaining_end,
                            start_hash: H256::zero(),
                            end_hash: None,
                        };
                        tasks_queue_not_started.push_back(task);
                        task_count += 1;
                    }
                    // Task found a big storage account, so we split the chunk into multiple chunks
                    let start_hash_u256 = U256::from_big_endian(&hash_start.0);
                    let missing_storage_range = U256::MAX - start_hash_u256;

                    // Big accounts need to be marked for storage healing unconditionally
                    for account in accounts_by_root_hash[remaining_start].1.iter() {
                        account_storage_roots.healed_accounts.insert(*account);
                    }

                    let slot_count = account_storages
                        .last()
                        .map(|v| v.len())
                        .ok_or(SnapError::NoAccountStorages)?
                        .max(1);
                    let storage_density = start_hash_u256 / slot_count;

                    let slots_per_chunk = U256::from(10000);
                    let chunk_size = storage_density
                        .checked_mul(slots_per_chunk)
                        .unwrap_or(U256::MAX);

                    let chunk_count = usize::try_from(missing_storage_range / chunk_size)
                        .unwrap_or(ACCOUNT_RANGE_CHUNK_COUNT)
                        .max(1);

                    let first_acc_hash = *accounts_by_root_hash[remaining_start]
                        .1
                        .first()
                        .ok_or(SnapError::InternalError("Empty accounts vector".to_owned()))?;

                    let maybe_old_intervals = account_storage_roots
                        .accounts_with_storage_root
                        .get(&first_acc_hash);

                    if let Some((_, old_intervals)) = maybe_old_intervals {
                        if !old_intervals.is_empty() {
                            for (start_hash, end_hash) in old_intervals {
                                let task = StorageTask {
                                    start_index: remaining_start,
                                    end_index: remaining_start + 1,
                                    start_hash: *start_hash,
                                    end_hash: Some(*end_hash),
                                };

                                tasks_queue_not_started.push_back(task);
                                task_count += 1;
                            }
                        } else {
                            // TODO: DRY
                            account_storage_roots
                                .accounts_with_storage_root
                                .insert(first_acc_hash, (None, vec![]));
                            let (_, intervals) = account_storage_roots
                                    .accounts_with_storage_root
                                    .get_mut(&first_acc_hash)
                                    .ok_or(SnapError::InternalError("Tried to get the old download intervals for an account but did not find them".to_owned()))?;

                            for i in 0..chunk_count {
                                let start_hash_u256 = start_hash_u256 + chunk_size * i;
                                let start_hash = H256::from_uint(&start_hash_u256);
                                let end_hash = if i == chunk_count - 1 {
                                    HASH_MAX
                                } else {
                                    let end_hash_u256 = start_hash_u256
                                        .checked_add(chunk_size)
                                        .unwrap_or(U256::MAX);
                                    H256::from_uint(&end_hash_u256)
                                };

                                let task = StorageTask {
                                    start_index: remaining_start,
                                    end_index: remaining_start + 1,
                                    start_hash,
                                    end_hash: Some(end_hash),
                                };

                                intervals.push((start_hash, end_hash));

                                tasks_queue_not_started.push_back(task);
                                task_count += 1;
                            }
                            debug!("Split big storage account into {chunk_count} chunks.");
                        }
                    } else {
                        account_storage_roots
                            .accounts_with_storage_root
                            .insert(first_acc_hash, (None, vec![]));
                        let (_, intervals) = account_storage_roots
                                .accounts_with_storage_root
                                .get_mut(&first_acc_hash)
                                .ok_or(SnapError::InternalError("Tried to get the old download intervals for an account but did not find them".to_owned()))?;

                        for i in 0..chunk_count {
                            let start_hash_u256 = start_hash_u256 + chunk_size * i;
                            let start_hash = H256::from_uint(&start_hash_u256);
                            let end_hash = if i == chunk_count - 1 {
                                HASH_MAX
                            } else {
                                let end_hash_u256 =
                                    start_hash_u256.checked_add(chunk_size).unwrap_or(U256::MAX);
                                H256::from_uint(&end_hash_u256)
                            };

                            let task = StorageTask {
                                start_index: remaining_start,
                                end_index: remaining_start + 1,
                                start_hash,
                                end_hash: Some(end_hash),
                            };

                            intervals.push((start_hash, end_hash));

                            tasks_queue_not_started.push_back(task);
                            task_count += 1;
                        }
                        debug!("Split big storage account into {chunk_count} chunks.");
                    }
                }
            } else if let Some(hash_end) = hash_end {
                // Per-interval task completed: the peer covered
                // [task_start_hash, hash_end] fully and verify_range reported
                // should_continue=false, so the worker returns
                // remaining_start == remaining_end and the guard above does
                // not fire. Drop the matching interval here so the account
                // can finalize across calls; otherwise the partition logic
                // at function entry would re-queue the same range forever.
                //
                // The helper returns false when no live interval is found —
                // that happens when a sibling per-interval task for the same
                // account already drained the last interval and finalized it
                // earlier in this call's loop. Unlike the partial-completion
                // path above (which panics on a missing acc_hash because no
                // sibling can have drained while work still remains in the
                // chunk), here we silently skip.
                clear_completed_interval(
                    account_storage_roots,
                    &accounts_by_root_hash,
                    &mut accounts_done,
                    start_index,
                    (task_start_hash, hash_end),
                )?;
            }

            if account_storages.is_empty() {
                peers.peer_table.record_failure(peer_id)?;
                continue;
            }
            if let Some(hash_end) = hash_end {
                // This is a big storage account, and the range might be empty
                if account_storages[0].len() == 1 && account_storages[0][0].0 > hash_end {
                    continue;
                }
            }

            peers.peer_table.record_success(peer_id)?;

            let n_storages = account_storages.len();
            let n_slots = account_storages
                .iter()
                .map(|storage| storage.len())
                .sum::<usize>();

            // These take into account we downloaded the same thing for different accounts
            let effective_slots: usize = account_storages
                .iter()
                .enumerate()
                .map(|(i, storages)| {
                    accounts_by_root_hash[start_index + i].1.len() * storages.len()
                })
                .sum();

            METRICS
                .storage_leaves_downloaded
                .inc_by(effective_slots as u64);

            debug!("Downloaded {n_storages} storages ({n_slots} slots) from peer {peer_id}");
            debug!(
                "Total tasks: {task_count}, completed tasks: {completed_tasks}, queued tasks: {}",
                tasks_queue_not_started.len()
            );
            // THEN: update insert to read with the correct structure and reuse
            // tries, only changing the prefix for insertion.
            if account_storages.len() == 1 {
                let (root_hash, accounts) = &accounts_by_root_hash[start_index];
                // We downloaded a big storage account
                current_account_storages
                    .entry(*root_hash)
                    .or_insert_with(|| AccountsWithStorage {
                        accounts: accounts.clone(),
                        storages: Vec::new(),
                    })
                    .storages
                    .extend(account_storages.remove(0));
            } else {
                for (i, storages) in account_storages.into_iter().enumerate() {
                    let (root_hash, accounts) = &accounts_by_root_hash[start_index + i];
                    current_account_storages.insert(
                        *root_hash,
                        AccountsWithStorage {
                            accounts: accounts.clone(),
                            storages,
                        },
                    );
                }
            }
        }

        if block_is_stale(pivot_header) {
            debug!("Pivot became stale during storage range download, stopping this round");
            break;
        }

        let Some((peer_id, connection, permit)) = peers
            .peer_table
            .get_best_peer(SUPPORTED_SNAP_CAPABILITIES.to_vec())
            .await
            .inspect_err(|err| warn!(%err, "Error requesting a peer for storage ranges"))
            .unwrap_or(None)
        else {
            // Log ~ once every 10 seconds
            if logged_no_free_peers_count == 0 {
                trace!("We are missing peers in request_storage_ranges");
                logged_no_free_peers_count = 1000;
            }
            logged_no_free_peers_count -= 1;
            // Sleep a bit to avoid busy polling
            tokio::time::sleep(Duration::from_millis(10)).await;
            continue;
        };

        let Some(task) = tasks_queue_not_started.pop_front() else {
            if completed_tasks >= task_count {
                break;
            }
            continue;
        };

        let tx = task_sender.clone();

        // FIXME: this unzip is probably pointless and takes up unnecessary memory.
        let (chunk_account_hashes, chunk_storage_roots): (Vec<_>, Vec<_>) = accounts_by_root_hash
            [task.start_index..task.end_index]
            .iter()
            .map(|(root, storages)| (*storages.first().unwrap_or(&H256::zero()), *root))
            .unzip();

        if task_count - completed_tasks < 30 {
            debug!(
                "Assigning task: {task:?}, account_hash: {}, storage_root: {}",
                chunk_account_hashes.first().unwrap_or(&H256::zero()),
                chunk_storage_roots.first().unwrap_or(&H256::zero()),
            );
        }
        tokio::spawn(request_storage_ranges_worker(
            task,
            peer_id,
            connection,
            pivot_header.state_root,
            chunk_account_hashes,
            chunk_storage_roots,
            tx,
            permit,
        ));
    }

    {
        let snapshot = current_account_storages.into_values().collect::<Vec<_>>();

        async_fs::ensure_dir_exists(account_storages_snapshots_dir).await?;

        let path = get_account_storages_snapshot_file(account_storages_snapshots_dir, chunk_index);
        dump_storages_to_file(&path, snapshot).map_err(|_| {
            SnapError::SnapshotDir(format!(
                "Failed to write storage snapshot chunk {}",
                chunk_index
            ))
        })?;
    }
    disk_joinset
        .join_all()
        .await
        .into_iter()
        .map(|result| {
            result.inspect_err(|err| error!("Failed to dump storage snapshot to file: {err:?}"))
        })
        .collect::<Result<Vec<()>, DumpError>>()
        .map_err(SnapError::from)?;

    for (account_done, intervals) in accounts_done {
        if intervals.is_empty() {
            account_storage_roots
                .accounts_with_storage_root
                .remove(&account_done);
        }
    }

    // Dropping the task sender so that the recv returns None
    drop(task_sender);

    Ok(chunk_index + 1)
}

/// Requests state trie nodes at the given paths from an already-selected peer.
/// Releases the peer slot as soon as the wire response is in; hash
/// verification below is pure computation.
/// Returns `SnapError::InvalidHash` if any returned node's hash does not match
/// the requested path, and `SnapError::InvalidData` on an empty or oversized
/// response.
pub async fn request_state_trienodes(
    mut connection: PeerConnection,
    permit: RequestPermit,
    state_root: H256,
    paths: Vec<RequestMetadata>,
) -> Result<Vec<Node>, SnapError> {
    let expected_nodes = paths.len();

    let request_id = rand::random();
    let request = RLPxMessage::GetTrieNodes(GetTrieNodes {
        id: request_id,
        root_hash: state_root,
        // [acc_path, acc_path,...] -> [[acc_path], [acc_path]]
        paths: paths
            .iter()
            .map(|vec| vec![Bytes::from(vec.path.encode_compact())])
            .collect(),
        bytes: MAX_RESPONSE_BYTES,
    });
    let response = connection
        .outgoing_request(request, PEER_REPLY_TIMEOUT)
        .await;
    drop(permit);
    let nodes = match response {
        Ok(RLPxMessage::TrieNodes(trie_nodes)) => trie_nodes
            .nodes
            .iter()
            .map(|node| Node::decode(node))
            .collect::<Result<Vec<_>, _>>()
            .map_err(SnapError::from),
        Ok(other_msg) => Err(SnapError::Protocol(
            PeerConnectionError::UnexpectedResponse("TrieNodes".to_string(), other_msg.to_string()),
        )),
        Err(other_err) => Err(SnapError::Protocol(other_err)),
    }?;

    if nodes.is_empty() || nodes.len() > expected_nodes {
        return Err(SnapError::InvalidData);
    }

    for (index, node) in nodes.iter().enumerate() {
        if node.compute_hash(&NativeCrypto).finalize(&NativeCrypto) != paths[index].hash {
            debug!(
                "A peer is sending wrong data for the state trie node {:?}",
                paths[index].path
            );
            return Err(SnapError::InvalidHash);
        }
    }

    Ok(nodes)
}

/// Requests storage trie nodes from an already-selected peer. The `GetTrieNodes`
/// payload carries the state root and the per-account paths (hashed address
/// prefix followed by storage-trie paths, which may be full or partial).
/// Consumes a `RequestPermit` reserved by the caller at peer selection time;
/// the permit drops when this function returns, releasing the slot.
/// Errors are returned as `RequestStorageTrieNodesError` carrying the
/// request ID so the caller can reconcile it with its in-flight map.
pub async fn request_storage_trienodes(
    mut connection: PeerConnection,
    _permit: RequestPermit,
    get_trie_nodes: GetTrieNodes,
) -> Result<TrieNodes, RequestStorageTrieNodesError> {
    let request_id = get_trie_nodes.id;
    let request = RLPxMessage::GetTrieNodes(get_trie_nodes);
    match connection
        .outgoing_request(request, PEER_REPLY_TIMEOUT)
        .await
    {
        Ok(RLPxMessage::TrieNodes(trie_nodes)) => Ok(trie_nodes),
        Ok(other_msg) => Err(RequestStorageTrieNodesError {
            request_id,
            source: SnapError::Protocol(PeerConnectionError::UnexpectedResponse(
                "TrieNodes".to_string(),
                other_msg.to_string(),
            )),
        }),
        Err(e) => Err(RequestStorageTrieNodesError {
            request_id,
            source: SnapError::Protocol(e),
        }),
    }
}

#[allow(clippy::type_complexity)]
async fn request_account_range_worker(
    peer_id: H256,
    mut connection: PeerConnection,
    chunk_start: H256,
    chunk_end: H256,
    state_root: H256,
    tx: tokio::sync::mpsc::Sender<(Vec<AccountRangeUnit>, H256, Option<(H256, H256)>)>,
    permit: RequestPermit,
) -> Result<(), SnapError> {
    debug!("Requesting account range from peer {peer_id}, chunk: {chunk_start:?} - {chunk_end:?}");
    let request_id = rand::random();
    let request = RLPxMessage::GetAccountRange(GetAccountRange {
        id: request_id,
        root_hash: state_root,
        starting_hash: chunk_start,
        limit_hash: chunk_end,
        response_bytes: MAX_RESPONSE_BYTES,
    });

    // Perform the wire request and release the peer slot as soon as the
    // response (or error) is in — processing below is pure computation.
    let response = connection
        .outgoing_request(request, PEER_REPLY_TIMEOUT)
        .await;
    drop(permit);

    let retry = || {
        (
            Vec::<AccountRangeUnit>::new(),
            Some((chunk_start, chunk_end)),
        )
    };
    let (accounts_out, chunk_left) = if let Ok(RLPxMessage::AccountRange(AccountRange {
        id: _,
        accounts,
        proof,
    })) = response
    {
        if accounts.is_empty() {
            retry()
        } else {
            // Validate response — build the verification inputs by borrowing
            // `accounts` so we can still consume it for the filtered output.
            let proof = encodable_to_proof(&proof);
            let account_hashes: Vec<H256> = accounts.iter().map(|u| u.hash).collect();
            let encoded_accounts: Vec<_> =
                accounts.iter().map(|u| u.account.encode_to_vec()).collect();

            match verify_range(
                state_root,
                &chunk_start,
                &account_hashes,
                &encoded_accounts,
                &proof,
            ) {
                Ok(should_continue) => {
                    let chunk_left = if should_continue {
                        match account_hashes.last() {
                            Some(last_hash) => {
                                let new_start_u256 = U256::from_big_endian(&last_hash.0) + 1;
                                let new_start = H256::from_uint(&new_start_u256);
                                Some((new_start, chunk_end))
                            }
                            None => {
                                // Unreachable: accounts is non-empty here.
                                error!("Account hashes last failed, this shouldn't happen");
                                Some((chunk_start, chunk_end))
                            }
                        }
                    } else {
                        None
                    };
                    let filtered = accounts
                        .into_iter()
                        .filter(|unit| unit.hash <= chunk_end)
                        .collect::<Vec<_>>();
                    (filtered, chunk_left)
                }
                Err(_) => {
                    tracing::debug!("Received invalid account range");
                    retry()
                }
            }
        }
    } else {
        tracing::debug!("Failed to get account range");
        retry()
    };

    tx.send((accounts_out, peer_id, chunk_left)).await.ok();
    Ok::<(), SnapError>(())
}

#[allow(clippy::too_many_arguments)]
async fn request_storage_ranges_worker(
    task: StorageTask,
    peer_id: H256,
    mut connection: PeerConnection,
    state_root: H256,
    chunk_account_hashes: Vec<H256>,
    chunk_storage_roots: Vec<H256>,
    tx: tokio::sync::mpsc::Sender<StorageTaskResult>,
    permit: RequestPermit,
) -> Result<(), SnapError> {
    let start = task.start_index;
    let end = task.end_index;
    let start_hash = task.start_hash;

    // Defaults for the "retry this same range" outcome used by every failure
    // branch below.
    let retry_outcome = || {
        (
            Vec::<Vec<(H256, U256)>>::new(),
            task.start_index,
            task.end_index,
            (start_hash, task.end_hash),
        )
    };

    let request_id = rand::random();
    let request = RLPxMessage::GetStorageRanges(GetStorageRanges {
        id: request_id,
        root_hash: state_root,
        account_hashes: chunk_account_hashes,
        starting_hash: start_hash,
        limit_hash: task.end_hash.unwrap_or(HASH_MAX),
        response_bytes: MAX_RESPONSE_BYTES,
    });
    tracing::trace!(peer_id = %peer_id, msg_type = "GetStorageRanges", "Sending storage range request");

    // Perform the wire request and release the peer slot as soon as the
    // response (or error) is in — validation below is pure computation.
    let response = connection
        .outgoing_request(request, PEER_REPLY_TIMEOUT)
        .await;
    drop(permit);

    let (account_storages, remaining_start, remaining_end, remaining_hash_range) = 'outcome: {
        let Ok(RLPxMessage::StorageRanges(StorageRanges {
            id: _,
            slots,
            proof,
        })) = response
        else {
            #[cfg(feature = "metrics")]
            ethrex_metrics::sync::METRICS_SYNC.inc_storage_request("timeout");
            tracing::trace!(peer_id = %peer_id, msg_type = "GetStorageRanges", outcome = "timeout", "Storage range request failed");
            tracing::debug!("Failed to get storage range");
            break 'outcome retry_outcome();
        };
        if slots.is_empty() && proof.is_empty() {
            #[cfg(feature = "metrics")]
            ethrex_metrics::sync::METRICS_SYNC.inc_storage_request("empty");
            tracing::trace!(peer_id = %peer_id, msg_type = "StorageRanges", outcome = "empty", "Storage range response empty");
            tracing::debug!("Received empty storage range");
            break 'outcome retry_outcome();
        }
        if slots.len() > chunk_storage_roots.len() || slots.is_empty() {
            break 'outcome retry_outcome();
        }
        let proof = encodable_to_proof(&proof);
        let mut account_storages: Vec<Vec<(H256, U256)>> = vec![];
        let mut should_continue = false;
        let mut validation_failed = false;
        let mut storage_roots = chunk_storage_roots.into_iter();
        let last_slot_index = slots.len() - 1;
        for (i, next_account_slots) in slots.into_iter().enumerate() {
            if next_account_slots.is_empty() {
                debug!("Received empty storage range, skipping");
                validation_failed = true;
                break;
            }
            let encoded_values = next_account_slots
                .iter()
                .map(|slot| slot.data.encode_to_vec())
                .collect::<Vec<_>>();
            let hashed_keys: Vec<_> = next_account_slots.iter().map(|slot| slot.hash).collect();

            let storage_root = match storage_roots.next() {
                Some(root) => root,
                None => {
                    debug!("No storage root for account {i}");
                    break 'outcome retry_outcome();
                }
            };

            // The proof corresponds to the last slot, for the previous ones the slot must be the full range without edge proofs
            if i == last_slot_index && !proof.is_empty() {
                let Ok(sc) = verify_range(
                    storage_root,
                    &start_hash,
                    &hashed_keys,
                    &encoded_values,
                    &proof,
                ) else {
                    validation_failed = true;
                    break;
                };
                should_continue = sc;
            } else if verify_range(
                storage_root,
                &start_hash,
                &hashed_keys,
                &encoded_values,
                &[],
            )
            .is_err()
            {
                validation_failed = true;
                break;
            }

            account_storages.push(
                next_account_slots
                    .iter()
                    .map(|slot| (slot.hash, slot.data))
                    .collect(),
            );
        }

        if validation_failed {
            break 'outcome retry_outcome();
        }

        let (remaining_start, remaining_end, remaining_start_hash) = if should_continue {
            let last_account_storage = match account_storages.last() {
                Some(storage) => storage,
                None => {
                    error!("No account storage found, this shouldn't happen");
                    break 'outcome retry_outcome();
                }
            };
            let (last_hash, _) = match last_account_storage.last() {
                Some(last_hash) => last_hash,
                None => {
                    error!("No last hash found, this shouldn't happen");
                    break 'outcome retry_outcome();
                }
            };
            let next_hash_u256 = U256::from_big_endian(&last_hash.0).saturating_add(1.into());
            let next_hash = H256::from_uint(&next_hash_u256);
            (start + account_storages.len() - 1, end, next_hash)
        } else {
            (start + account_storages.len(), end, H256::zero())
        };
        let slot_count: usize = account_storages.iter().map(|s| s.len()).sum();
        #[cfg(feature = "metrics")]
        ethrex_metrics::sync::METRICS_SYNC.inc_storage_request("success");
        tracing::trace!(peer_id = %peer_id, msg_type = "StorageRanges", outcome = "success", slots = slot_count, "Storage range response received");
        (
            account_storages,
            remaining_start,
            remaining_end,
            (remaining_start_hash, task.end_hash),
        )
    };

    let task_result = StorageTaskResult {
        start_index: start,
        account_storages,
        peer_id,
        remaining_start,
        remaining_end,
        remaining_hash_range,
        task_start_hash: start_hash,
    };
    tx.send(task_result).await.ok();
    Ok::<(), SnapError>(())
}