fixes

tests/compat/src/cli.rs

@@ -17,15 +17,25 @@ pub struct Args {
     /// Base URL to use for `monerod` RPC.
     ///
     /// This must be a non-restricted RPC.
-    #[arg(short, long, default_value_t = String::from("http://127.0.0.1:18081"))]
+    #[arg(long, default_value_t = String::from("http://127.0.0.1:18081"))]
     pub rpc_url: String,
 
+    /// Amount of async RPC tasks to spawn.
+    #[arg(long, default_value_t = NonZeroUsize::new(4).unwrap())]
+    pub rpc_tasks: NonZeroUsize,
+
+    /// The maximum capacity of the block buffer in-between the RPC and verifier.
+    ///
+    /// `0` will cause the buffer to be unbounded.
+    #[arg(long, default_value_t = 1000)]
+    pub buffer_limit: usize,
+
     /// Amount of verifying threads to spawn.
-    #[arg(short, long, default_value_t = std::thread::available_parallelism().unwrap())]
+    #[arg(long, default_value_t = std::thread::available_parallelism().unwrap())]
     pub threads: NonZeroUsize,
 
     /// Print an update every `update` amount of blocks.
-    #[arg(short, long, default_value_t = NonZeroU64::new(500).unwrap())]
+    #[arg(long, default_value_t = NonZeroU64::new(500).unwrap())]
     pub update: NonZeroU64,
 }
 

tests/compat/src/main.rs

@@ -21,17 +21,23 @@ async fn main() {
     let cli::Args {
         rpc_url,
         update,
+        rpc_tasks,
+        buffer_limit,
         threads,
     } = cli::Args::get();
 
     // Set-up RPC client.
-    let client = rpc::RpcClient::new(rpc_url).await;
+    let client = rpc::RpcClient::new(rpc_url, rpc_tasks).await;
     let top_height = client.top_height;
     println!("top_height: {top_height}");
     println!();
 
     // Test.
-    let (tx, rx) = crossbeam::channel::unbounded();
+    let (tx, rx) = if buffer_limit == 0 {
+        crossbeam::channel::unbounded()
+    } else {
+        crossbeam::channel::bounded(buffer_limit)
+    };
     verify::spawn_verify_pool(threads, update, top_height, rx);
     client.test(top_height, tx).await;
 

tests/compat/src/rpc.rs

@@ -1,4 +1,4 @@
-use std::time::Instant;
+use std::{num::NonZeroUsize, time::Instant};
 
 use crossbeam::channel::Sender;
 use monero_serai::{block::Block, transaction::Transaction};
@@ -20,11 +20,12 @@ pub struct RpcClient {
     client: Client,
     json_rpc_url: String,
     get_transactions_url: String,
+    rpc_tasks: NonZeroUsize,
     pub top_height: u64,
 }
 
 impl RpcClient {
-    pub async fn new(rpc_url: String) -> Self {
+    pub async fn new(rpc_url: String, rpc_tasks: NonZeroUsize) -> Self {
         let headers = {
             let mut h = HeaderMap::new();
             h.insert("Content-Type", HeaderValue::from_static("application/json"));
@@ -70,6 +71,7 @@ impl RpcClient {
             client,
             json_rpc_url,
             get_transactions_url,
+            rpc_tasks,
             top_height,
         }
     }
@@ -207,7 +209,7 @@ impl RpcClient {
         });
 
         futures::stream::iter(iter)
-            .buffer_unordered(4) // This can't be too high or else we get bottlenecked by `monerod`
+            .buffer_unordered(self.rpc_tasks.get()) // This can't be too high or else we get bottlenecked by `monerod`
             .for_each(|()| async {})
             .await;
     }

tests/compat/src/verify.rs

@@ -17,7 +17,7 @@ use crate::{
 
 struct Verifier {
     id: usize,
-    now: Instant,
+    start_of_new_pow: Instant,
     thread_count: NonZeroUsize,
     update: NonZeroU64,
     top_height: u64,
@@ -33,14 +33,12 @@ pub fn spawn_verify_pool(
     top_height: u64,
     rx: Receiver<RpcBlockData>,
 ) {
-    let now = Instant::now();
+    for id in 1..=thread_count.get() {
-
-    for id in 0..thread_count.get() {
         let rx = rx.clone();
         std::thread::spawn(move || {
             Verifier {
                 id,
-                now,
+                start_of_new_pow: Instant::now(),
                 thread_count,
                 update,
                 top_height,
@@ -80,39 +78,14 @@ impl Verifier {
         } = data;
         let GetBlockResponse { blob, block_header } = get_block_response;
 
-        //----------------------------------------------- Calculate some data.
-        let calculated_block_reward = block
-            .miner_transaction
-            .prefix()
-            .outputs
-            .iter()
-            .map(|o| o.amount.unwrap())
-            .sum::<u64>();
-        let calculated_block_weight = txs
-            .iter()
-            .map(|RpcTxData { tx, .. }| tx.weight())
-            .sum::<usize>();
-        let calculated_pow_data = block.serialize_pow_hash();
-        let miner_tx_weight = block.miner_transaction.weight();
-
         //----------------------------------------------- Verify.
-        Self::verify_block_properties(&blob, &block, calculated_block_reward, &p);
+        Self::verify_blocks(&blob, &block, &txs, &p, block_header);
-        Self::verify_all_transactions_are_unique(&txs, &p);
+        Self::verify_transactions(txs, &p);
-        Self::verify_transaction_properties(txs, &p);
-
-        Self::verify_block_fields(
-            calculated_block_weight,
-            calculated_block_reward,
-            &block,
-            &p,
-            block_header,
-        );
-
         let algo = self.verify_pow(
             block_header.height,
             seed_hash,
             block_header.pow_hash,
-            &calculated_pow_data,
+            &block.serialize_pow_hash(),
             &p,
         );
 
@@ -120,27 +93,79 @@ impl Verifier {
         self.print_progress(
             algo,
             seed_height,
-            miner_tx_weight,
+            block.miner_transaction.weight(),
             blocks_per_sec,
             block_header,
         );
     }
 
-    fn verify_block_properties(
+    fn verify_blocks(
         block_blob: &[u8],
         block: &Block,
-        calculated_block_reward: u64,
+        txs: &[RpcTxData],
         p: &str,
+        BlockHeader {
+            block_weight,
+            hash,
+            pow_hash: _,
+            height,
+            major_version,
+            minor_version,
+            miner_tx_hash,
+            nonce,
+            num_txes,
+            prev_hash,
+            reward,
+            timestamp,
+        }: BlockHeader,
     ) {
-        // Test block properties.
+        let calculated_block_reward = block
+            .miner_transaction
+            .prefix()
+            .outputs
+            .iter()
+            .map(|o| o.amount.unwrap())
+            .sum::<u64>();
+
+        let calculated_block_weight = txs
+            .iter()
+            .map(|RpcTxData { tx, .. }| tx.weight())
+            .sum::<usize>();
+
+        let block_number = u64::try_from(block.number().unwrap()).unwrap();
+
+        assert!(!block.miner_transaction.prefix().outputs.is_empty(), "{p}");
+        assert_ne!(calculated_block_reward, 0, "{p}");
+        assert_ne!(block.miner_transaction.weight(), 0, "{p}");
         assert_eq!(block_blob, block.serialize(), "{p}");
+        assert_eq!(block_weight, calculated_block_weight, "{p}");
+        assert_eq!(hash, block.hash(), "{p}");
+        assert_eq!(height, block_number, "{p}");
+        assert_eq!(major_version, block.header.hardfork_version, "{p}");
+        assert_eq!(minor_version, block.header.hardfork_signal, "{p}");
+        assert_eq!(miner_tx_hash, block.miner_transaction.hash(), "{p}");
+        assert_eq!(nonce, block.header.nonce, "{p}");
+        assert_eq!(num_txes, block.transactions.len(), "{p}");
+        assert_eq!(prev_hash, block.header.previous, "{p}");
+        assert_eq!(reward, calculated_block_reward, "{p}");
+        assert_eq!(timestamp, block.header.timestamp, "{p}");
+    }
 
-        assert!(
+    fn verify_transactions(txs: Vec<RpcTxData>, p: &str) {
-            !block.miner_transaction.prefix().outputs.is_empty(),
+        Self::verify_all_transactions_are_unique(&txs, p);
-            "miner_tx has no outputs\n{p}"
-        );
 
-        assert_ne!(calculated_block_reward, 0, "block reward is 0\n{p}");
+        for RpcTxData {
+            tx,
+            tx_blob,
+            tx_hash,
+        } in txs
+        {
+            assert_eq!(tx_hash, tx.hash(), "{p}, tx: {tx:#?}");
+            assert_ne!(tx.weight(), 0, "{p}, tx: {tx:#?}");
+            assert!(!tx.prefix().inputs.is_empty(), "{p}, tx: {tx:#?}");
+            assert_eq!(tx_blob, tx.serialize(), "{p}, tx: {tx:#?}");
+            assert!(matches!(tx.version(), 1 | 2), "{p}, tx: {tx:#?}");
+        }
     }
 
     #[expect(clippy::significant_drop_tightening)]
@@ -159,61 +184,6 @@ impl Verifier {
         }
     }
 
-    fn verify_transaction_properties(txs: Vec<RpcTxData>, p: &str) {
-        // Test transaction properties.
-        for RpcTxData {
-            tx,
-            tx_blob,
-            tx_hash,
-        } in txs
-        {
-            assert_eq!(tx_hash, tx.hash(), "{p}, tx: {tx:#?}");
-            assert_ne!(tx.weight(), 0, "{p}, tx: {tx:#?}");
-            assert!(!tx.prefix().inputs.is_empty(), "{p}, tx: {tx:#?}");
-            assert_eq!(tx_blob, tx.serialize(), "{p}, tx: {tx:#?}");
-            assert!(matches!(tx.version(), 1 | 2), "{p}, tx: {tx:#?}");
-        }
-    }
-
-    fn verify_block_fields(
-        calculated_block_weight: usize,
-        calculated_block_reward: u64,
-        block: &Block,
-        p: &str,
-        BlockHeader {
-            block_weight,
-            hash,
-            pow_hash: _,
-            height,
-            major_version,
-            minor_version,
-            miner_tx_hash,
-            nonce,
-            num_txes,
-            prev_hash,
-            reward,
-            timestamp,
-        }: BlockHeader,
-    ) {
-        // Test block fields are correct.
-        assert_eq!(block_weight, calculated_block_weight, "{p}");
-        assert_ne!(block.miner_transaction.weight(), 0, "{p}");
-        assert_eq!(hash, block.hash(), "{p}");
-        assert_eq!(
-            height,
-            u64::try_from(block.number().unwrap()).unwrap(),
-            "{p}"
-        );
-        assert_eq!(major_version, block.header.hardfork_version, "{p}");
-        assert_eq!(minor_version, block.header.hardfork_signal, "{p}");
-        assert_eq!(miner_tx_hash, block.miner_transaction.hash(), "{p}");
-        assert_eq!(nonce, block.header.nonce, "{p}");
-        assert_eq!(num_txes, block.transactions.len(), "{p}");
-        assert_eq!(prev_hash, block.header.previous, "{p}");
-        assert_eq!(reward, calculated_block_reward, "{p}");
-        assert_eq!(timestamp, block.header.timestamp, "{p}");
-    }
-
     fn verify_pow(
         &mut self,
         height: u64,
@@ -222,6 +192,10 @@ impl Verifier {
         calculated_pow_data: &[u8],
         p: &str,
     ) -> &'static str {
+        if matches!(height, 1546000 | 1685555 | 1788000 | 1978433) {
+            self.start_of_new_pow = Instant::now();
+        }
+
         let (algo, calculated_pow_hash) = if height < RANDOMX_START_HEIGHT {
             CryptoNightHash::hash(calculated_pow_data, height)
         } else {
@@ -301,7 +275,7 @@ impl Verifier {
         let relative_count = find_count_relative_to_pow_activation(height);
         let block_buffer = self.rx.len();
 
-        let elapsed = self.now.elapsed().as_secs_f64();
+        let elapsed = self.start_of_new_pow.elapsed().as_secs_f64();
         let secs_per_hash = elapsed / relative_count as f64;
         let verify_bps = relative_count as f64 / elapsed;
         let remaining_secs = (top_height as f64 - relative_count as f64) * secs_per_hash;