cuprate_consensus/block/
batch_prepare.rs

1use std::{collections::HashMap, sync::Arc};
2
3use monero_oxide::{block::Block, transaction::Transaction};
4use rayon::prelude::*;
5use tower::{Service, ServiceExt};
6use tracing::instrument;
7
8use cuprate_consensus_context::{rx_vms::RandomXVm, BlockchainContextService};
9use cuprate_consensus_rules::{
10    blocks::{check_block_pow, is_randomx_seed_height, randomx_seed_height, BlockError},
11    hard_forks::HardForkError,
12    miner_tx::MinerTxError,
13    ConsensusError, HardFork,
14};
15use cuprate_helper::asynch::rayon_spawn_async;
16use cuprate_types::{output_cache::OutputCache, TransactionVerificationData};
17
18use crate::{
19    __private::Database,
20    batch_verifier::MultiThreadedBatchVerifier,
21    block::{free::order_transactions, PreparedBlock, PreparedBlockExPow},
22    transactions::{check_kis_unique, contextual_data::get_output_cache, start_tx_verification},
23    BlockChainContextRequest, BlockChainContextResponse, ExtendedConsensusError,
24};
25
26/// Cached state created when batch preparing a group of blocks.
27///
28/// This cache is only valid for the set of blocks it was created with, it should not be used for
29/// other blocks.
30pub struct BatchPrepareCache {
31    pub(crate) output_cache: OutputCache,
32}
33
34/// Batch prepares a list of blocks for verification.
35#[instrument(level = "debug", name = "batch_prep_blocks", skip_all, fields(amt = blocks.len()))]
36#[expect(clippy::type_complexity)]
37pub async fn batch_prepare_main_chain_blocks<D: Database>(
38    blocks: Vec<(Block, Vec<Transaction>)>,
39    context_svc: &mut BlockchainContextService,
40    mut database: D,
41) -> Result<
42    (
43        Vec<(PreparedBlock, Vec<TransactionVerificationData>)>,
44        BatchPrepareCache,
45    ),
46    ExtendedConsensusError,
47> {
48    let (blocks, txs): (Vec<_>, Vec<_>) = blocks.into_iter().unzip();
49
50    tracing::debug!("Calculating block hashes.");
51    let blocks: Vec<PreparedBlockExPow> = rayon_spawn_async(|| {
52        blocks
53            .into_iter()
54            .map(PreparedBlockExPow::new)
55            .collect::<Result<Vec<_>, _>>()
56    })
57    .await?;
58
59    let Some(last_block) = blocks.last() else {
60        return Err(ConsensusError::Block(BlockError::NoBlocksToVerify).into());
61    };
62
63    // hard-forks cannot be reversed, so the last block will contain the highest hard fork (provided the
64    // batch is valid).
65    let top_hf_in_batch = last_block.hf_version;
66
67    // A Vec of (timestamp, HF) for each block to calculate the expected difficulty for each block.
68    let mut timestamps_hfs = Vec::with_capacity(blocks.len());
69    let mut new_rx_vm = None;
70
71    tracing::debug!("Checking blocks follow each other.");
72
73    // For every block make sure they have the correct height and previous ID
74    for window in blocks.windows(2) {
75        let block_0 = &window[0];
76        let block_1 = &window[1];
77
78        // Make sure no blocks in the batch have a higher hard fork than the last block.
79        if block_0.hf_version > top_hf_in_batch {
80            return Err(ConsensusError::Block(BlockError::HardForkError(
81                HardForkError::VersionIncorrect,
82            ))
83            .into());
84        }
85
86        if block_0.block_hash != block_1.block.header.previous
87            || block_0.height != block_1.height - 1
88        {
89            tracing::debug!("Blocks do not follow each other, verification failed.");
90            return Err(ConsensusError::Block(BlockError::PreviousIDIncorrect).into());
91        }
92
93        // Cache any potential RX VM seeds as we may need them for future blocks in the batch.
94        if is_randomx_seed_height(block_0.height) && top_hf_in_batch >= HardFork::V12 {
95            new_rx_vm = Some((block_0.height, block_0.block_hash));
96        }
97
98        timestamps_hfs.push((block_0.block.header.timestamp, block_0.hf_version));
99    }
100
101    // Calculate the expected difficulties for each block in the batch.
102    let BlockChainContextResponse::BatchDifficulties(difficulties) = context_svc
103        .ready()
104        .await?
105        .call(BlockChainContextRequest::BatchGetDifficulties(
106            timestamps_hfs,
107        ))
108        .await?
109    else {
110        panic!("Context service returned wrong response!");
111    };
112
113    // Get the current blockchain context.
114    let context = context_svc.blockchain_context();
115
116    // Make sure the blocks follow the main chain.
117
118    if context.chain_height != blocks[0].height {
119        tracing::debug!("Blocks do not follow main chain, verification failed.");
120
121        return Err(ConsensusError::Block(BlockError::MinerTxError(
122            MinerTxError::InputsHeightIncorrect,
123        ))
124        .into());
125    }
126
127    if context.top_hash != blocks[0].block.header.previous {
128        tracing::debug!("Blocks do not follow main chain, verification failed.");
129
130        return Err(ConsensusError::Block(BlockError::PreviousIDIncorrect).into());
131    }
132
133    let mut rx_vms = if top_hf_in_batch < HardFork::V12 {
134        HashMap::new()
135    } else {
136        let BlockChainContextResponse::RxVms(rx_vms) = context_svc
137            .ready()
138            .await?
139            .call(BlockChainContextRequest::CurrentRxVms)
140            .await?
141        else {
142            panic!("Blockchain context service returned wrong response!");
143        };
144
145        rx_vms
146    };
147
148    // If we have a RX seed in the batch calculate it.
149    if let Some((new_vm_height, new_vm_seed)) = new_rx_vm {
150        tracing::debug!("New randomX seed in batch, initialising VM");
151
152        let new_vm = rayon_spawn_async(move || {
153            Arc::new(RandomXVm::new(&new_vm_seed).expect("RandomX VM gave an error on set up!"))
154        })
155        .await;
156
157        // Give the new VM to the context service, so it can cache it.
158        context_svc
159            .oneshot(BlockChainContextRequest::NewRXVM((
160                new_vm_seed,
161                Arc::clone(&new_vm),
162            )))
163            .await?;
164
165        rx_vms.insert(new_vm_height, new_vm);
166    }
167
168    tracing::debug!("Calculating PoW and prepping transaction");
169
170    let blocks = rayon_spawn_async(move || {
171        let batch_verifier = MultiThreadedBatchVerifier::new(rayon::current_num_threads());
172
173        let res = blocks
174            .into_par_iter()
175            .zip(difficulties)
176            .zip(txs)
177            .map(|((block, difficulty), txs)| {
178                // Calculate the PoW for the block.
179                let height = block.height;
180                let block = PreparedBlock::new_prepped(
181                    block,
182                    rx_vms.get(&randomx_seed_height(height)).map(AsRef::as_ref),
183                )?;
184
185                // Check the PoW
186                check_block_pow(&block.pow_hash, difficulty).map_err(ConsensusError::Block)?;
187
188                let mut txs = start_tx_verification()
189                    .append_txs(txs)
190                    .prepare()?
191                    .only_semantic(block.hf_version)
192                    .queue(&batch_verifier)?;
193
194                // Order the txs correctly.
195                order_transactions(&block.block, &mut txs)?;
196
197                Ok((block, txs))
198            })
199            .collect::<Result<Vec<_>, ConsensusError>>()?;
200
201        if !batch_verifier.verify() {
202            return Err(ConsensusError::Block(BlockError::BatchVerificationFailed));
203        }
204
205        Ok(res)
206    })
207    .await?;
208
209    check_kis_unique(blocks.iter().flat_map(|(_, txs)| txs.iter()), &mut database).await?;
210
211    let output_cache =
212        get_output_cache(blocks.iter().flat_map(|(_, txs)| txs.iter()), database).await?;
213
214    Ok((blocks, BatchPrepareCache { output_cache }))
215}
cuprate_consensus/block/batch_prepare.rs

cuprate_consensus/block/
batch_prepare.rs
