cuprate_fast_sync/

fast_sync.rs

use std::{
    cmp::min,
    collections::{HashMap, VecDeque},
};

use blake3::Hasher;
use monero_oxide::{
    block::Block,
    transaction::{Input, Transaction},
};
use tower::{Service, ServiceExt};

use cuprate_blockchain::service::BlockchainReadHandle;
use cuprate_consensus::transactions::new_tx_verification_data;
use cuprate_consensus_context::BlockchainContext;
use cuprate_p2p::block_downloader::ChainEntry;
use cuprate_p2p_core::NetworkZone;
use cuprate_types::{
    blockchain::{BlockchainReadRequest, BlockchainResponse},
    Chain, VerifiedBlockInformation, VerifiedTransactionInformation,
};

/// The size of a batch of block hashes to hash to create a fast sync hash.
pub const FAST_SYNC_BATCH_LEN: usize = 512;

/// Returns the height of the first block not included in the given hashes.
pub const fn fast_sync_stop_height(hashes: &[[u8; 32]]) -> usize {
    hashes.len() * FAST_SYNC_BATCH_LEN
}

/// Validates that the given [`ChainEntry`]s are in the fast-sync hashes.
///
/// `entries` should be a list of sequential entries.
/// `start_height` should be the height of the first block in the first entry.
/// `fast_sync_hashes` should be the hashes to validate against (empty to skip validation).
///
/// Returns a tuple, the first element being the entries that are valid* the second
/// the entries we do not know are valid and should be passed in again when we have more entries.
///
/// *once we are passed the fast sync blocks all entries will be returned as valid as
/// we can not check their validity here.
///
/// There may be more entries returned than passed in as entries could be split.
pub async fn validate_entries<N: NetworkZone>(
    mut entries: VecDeque<ChainEntry<N>>,
    start_height: usize,
    blockchain_read_handle: &mut BlockchainReadHandle,
    fast_sync_hashes: &[[u8; 32]],
) -> Result<(VecDeque<ChainEntry<N>>, VecDeque<ChainEntry<N>>), tower::BoxError> {
    let stop_height = fast_sync_stop_height(fast_sync_hashes);

    // if we are past the top fast sync block return all entries as valid.
    if start_height >= stop_height {
        return Ok((entries, VecDeque::new()));
    }

    /*
       The algorithm used here needs to preserve which peer told us about which blocks, so we cannot
       simply join all the hashes together return all the ones that can be validated and the ones that
       can't, we need to keep the batches separate.

       The first step is to calculate how many hashes we need from the blockchain to make up the first
       fast-sync hash.

       Then will take out all the batches at the end for which we cannot make up a full fast-sync hash
       for, we will split a batch if it can only be partially validated.

       With the remaining hashes from the blockchain and the hashes in the batches we can validate we
       work on calculating the fast sync hashes and comparing them to the ones provided.
    */

    // First calculate the start and stop for this range of hashes.
    let hashes_start_height = (start_height / FAST_SYNC_BATCH_LEN) * FAST_SYNC_BATCH_LEN;
    let amount_of_hashes = entries.iter().map(|e| e.ids.len()).sum::<usize>();
    let last_height = amount_of_hashes + start_height;

    let hashes_stop_height = min(
        (last_height / FAST_SYNC_BATCH_LEN) * FAST_SYNC_BATCH_LEN,
        stop_height,
    );

    let mut hashes_stop_diff_last_height = last_height - hashes_stop_height;

    let starting_hashes = if hashes_start_height == start_height {
        vec![]
    } else {
        // get the hashes we are missing to create the first fast-sync hash.
        let BlockchainResponse::BlockHashInRange(starting_hashes) = blockchain_read_handle
            .ready()
            .await?
            .call(BlockchainReadRequest::BlockHashInRange(
                hashes_start_height..start_height,
                Chain::Main,
            ))
            .await?
        else {
            unreachable!()
        };

        starting_hashes
    };

    // If we don't have enough hashes to make up a batch we can't validate any.
    if amount_of_hashes + starting_hashes.len() < FAST_SYNC_BATCH_LEN {
        return Ok((VecDeque::new(), entries));
    }

    let mut unknown = VecDeque::new();

    // start moving from the back of the batches taking enough hashes out so we are only left with hashes
    // that can be verified.
    while !entries.is_empty() && hashes_stop_diff_last_height != 0 {
        let back = entries.back_mut().unwrap();

        if back.ids.len() >= hashes_stop_diff_last_height {
            // This batch is partially valid so split it.
            unknown.push_front(ChainEntry {
                ids: back
                    .ids
                    .drain((back.ids.len() - hashes_stop_diff_last_height)..)
                    .collect(),
                peer: back.peer,
                handle: back.handle.clone(),
            });

            break;
        }

        // Add this batch to the front of the unknowns, we do not know its validity.
        let back = entries.pop_back().unwrap();
        hashes_stop_diff_last_height -= back.ids.len();
        unknown.push_front(back);
    }

    // Start verifying the hashes.
    let mut hasher = Hasher::default();
    let mut last_i = 1;
    for (i, hash) in starting_hashes
        .iter()
        .chain(entries.iter().flat_map(|e| e.ids.iter()))
        .enumerate()
    {
        hasher.update(hash);

        if (i + 1) % FAST_SYNC_BATCH_LEN == 0 {
            let got_hash = hasher.finalize();

            if got_hash != fast_sync_hashes[get_hash_index_for_height(hashes_start_height + i)] {
                return Err("Hashes do not match".into());
            }
            hasher.reset();
        }

        last_i = i + 1;
    }
    // Make sure we actually checked all hashes.
    assert_eq!(last_i % FAST_SYNC_BATCH_LEN, 0);

    Ok((entries, unknown))
}

/// Get the index of the hash that contains this block in the fast sync hashes.
const fn get_hash_index_for_height(height: usize) -> usize {
    height / FAST_SYNC_BATCH_LEN
}

/// Creates a [`VerifiedBlockInformation`] from a block known to be valid.
///
/// # Panics
///
/// This may panic if used on an invalid block.
pub fn block_to_verified_block_information(
    block: Block,
    txs: Vec<Transaction>,
    blockchin_ctx: &BlockchainContext,
) -> VerifiedBlockInformation {
    let block_hash = block.hash();

    let block_blob = block.serialize();

    let Some(Input::Gen(height)) = block.miner_transaction().prefix().inputs.first() else {
        panic!("fast sync block invalid");
    };

    assert_eq!(
        *height, blockchin_ctx.chain_height,
        "fast sync block invalid"
    );

    let mut txs = txs
        .into_iter()
        .map(|tx| {
            let data = new_tx_verification_data(tx).expect("fast sync block invalid");

            (data.tx_hash, data)
        })
        .collect::<HashMap<_, _>>();

    let mut verified_txs = Vec::with_capacity(txs.len());
    for tx in &block.transactions {
        let data = txs.remove(tx).expect("fast sync block invalid");

        let (tx, prunable) = data.tx.pruned_with_prunable();
        verified_txs.push(VerifiedTransactionInformation {
            tx_prunable_blob: prunable,
            tx_pruned: tx.serialize(),
            tx_weight: data.tx_weight,
            fee: data.fee,
            tx_hash: data.tx_hash,
            tx,
        });
    }

    let total_fees = verified_txs.iter().map(|tx| tx.fee).sum::<u64>();
    let total_outputs = block
        .miner_transaction()
        .prefix()
        .outputs
        .iter()
        .map(|output| output.amount.unwrap_or(0))
        .sum::<u64>();

    let generated_coins = total_outputs - total_fees;

    let weight = block.miner_transaction().weight()
        + verified_txs.iter().map(|tx| tx.tx_weight).sum::<usize>();

    VerifiedBlockInformation {
        block_blob,
        txs: verified_txs,
        block_hash,
        pow_hash: [u8::MAX; 32],
        height: *height,
        generated_coins,
        weight,
        long_term_weight: blockchin_ctx.next_block_long_term_weight(weight),
        cumulative_difficulty: blockchin_ctx.cumulative_difficulty + blockchin_ctx.next_difficulty,
        block,
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::VecDeque,
        path::PathBuf,
        slice,
        sync::{Arc, LazyLock},
    };

    use proptest::proptest;

    use cuprate_blockchain::{config::Config, service::BlockchainReadHandle};
    use cuprate_p2p::block_downloader::ChainEntry;
    use cuprate_p2p_core::{client::InternalPeerID, handles::HandleBuilder, ClearNet};

    use crate::{fast_sync_stop_height, validate_entries, FAST_SYNC_BATCH_LEN};

    static HASHES: LazyLock<&[[u8; 32]]> = LazyLock::new(|| {
        (0..FAST_SYNC_BATCH_LEN * 2000)
            .map(|i| {
                let mut ret = [0; 32];
                ret[..8].copy_from_slice(&i.to_le_bytes());
                ret
            })
            .collect::<Vec<_>>()
            .leak()
    });

    static FAST_SYNC_HASHES: LazyLock<&[[u8; 32]]> = LazyLock::new(|| {
        HASHES
            .chunks(FAST_SYNC_BATCH_LEN)
            .map(|chunk| {
                let len = chunk.len() * 32;
                let bytes = chunk.as_ptr().cast::<u8>();

                // SAFETY:
                // We are casting a valid [[u8; 32]] to a [u8], no alignment requirements and we are using it
                // within the [[u8; 32]]'s lifetime.
                unsafe { blake3::hash(slice::from_raw_parts(bytes, len)).into() }
            })
            .collect::<Vec<_>>()
            .leak()
    });

    fn test_db(path: PathBuf) -> BlockchainReadHandle {
        let config = Config {
            blob_dir: path.clone(),
            index_dir: path.clone(),
            ..Default::default()
        };

        let fjall = fjall::Database::builder(path).open().unwrap();

        let thread_pool = Arc::new(rayon::ThreadPoolBuilder::new().build().unwrap());

        let (blockchain_read_handle, _, _) =
            cuprate_blockchain::service::init_with_pool(&config, fjall, thread_pool).unwrap();

        blockchain_read_handle
    }

    proptest! {
        #[test]
        fn valid_entry(len in 0_usize..1_500_000) {
            let mut ids = HASHES.to_vec();
            ids.resize(len, [0_u8; 32]);

            let handle = HandleBuilder::new().build();

            let entry = ChainEntry {
                ids,
                peer: InternalPeerID::Unknown([1; 16]),
                handle: handle.1
            };

            let data_dir = tempfile::tempdir().unwrap();

            tokio_test::block_on(async move {
                let mut blockchain_read_handle= test_db(data_dir.path().to_path_buf());

                let ret = validate_entries::<ClearNet>(VecDeque::from([entry]), 0, &mut blockchain_read_handle, *FAST_SYNC_HASHES).await.unwrap();

                let len_left = ret.0.iter().map(|e| e.ids.len()).sum::<usize>();
                let len_right = ret.1.iter().map(|e| e.ids.len()).sum::<usize>();

                assert_eq!(len_left + len_right, len);
                assert!(len_left <= fast_sync_stop_height(*FAST_SYNC_HASHES));
                assert!(len_right < FAST_SYNC_BATCH_LEN || len > fast_sync_stop_height(*FAST_SYNC_HASHES));
            });
        }

        #[test]
        fn single_hash_entries(len in 0_usize..1_500_000) {
            let handle = HandleBuilder::new().build();
            let entries = (0..len).map(|i| {
                ChainEntry {
                    ids: vec![HASHES.get(i).copied().unwrap_or_default()],
                    peer: InternalPeerID::Unknown([1; 16]),
                    handle: handle.1.clone()
                }
            }).collect();

            let data_dir = tempfile::tempdir().unwrap();

            tokio_test::block_on(async move {
                let mut blockchain_read_handle= test_db(data_dir.path().to_path_buf());

                let ret = validate_entries::<ClearNet>(entries, 0, &mut blockchain_read_handle, *FAST_SYNC_HASHES).await.unwrap();

                let len_left = ret.0.iter().map(|e| e.ids.len()).sum::<usize>();
                let len_right = ret.1.iter().map(|e| e.ids.len()).sum::<usize>();

                assert_eq!(len_left + len_right, len);
                assert!(len_left <= fast_sync_stop_height(*FAST_SYNC_HASHES));
                assert!(len_right < FAST_SYNC_BATCH_LEN || len > fast_sync_stop_height(*FAST_SYNC_HASHES));
            });
        }

        #[test]
        fn not_enough_hashes(len in 0_usize..FAST_SYNC_BATCH_LEN) {
            let hashes_start_height = FAST_SYNC_BATCH_LEN * 1234;

            let handle = HandleBuilder::new().build();
            let entry = ChainEntry {
                ids: HASHES[hashes_start_height..(hashes_start_height + len)].to_vec(),
                peer: InternalPeerID::Unknown([1; 16]),
                handle: handle.1
            };

            let data_dir = tempfile::tempdir().unwrap();

            tokio_test::block_on(async move {
                let mut blockchain_read_handle= test_db(data_dir.path().to_path_buf());

                let ret = validate_entries::<ClearNet>(VecDeque::from([entry]), 0, &mut blockchain_read_handle, *FAST_SYNC_HASHES).await.unwrap();

                let len_left = ret.0.iter().map(|e| e.ids.len()).sum::<usize>();
                let len_right = ret.1.iter().map(|e| e.ids.len()).sum::<usize>();

                assert_eq!(len_right, len);
                assert_eq!(len_left, 0);
            });
        }
    }
}