feature: Check the previous block height in BlockVerifier

This is a temporary busy-waiting fix.

zebra-consensus/Cargo.toml

@@ -7,22 +7,23 @@ edition = "2018"
 
 [dependencies]
 chrono = "0.4.13"
-futures = "0.3.5"
+color-eyre = "0.5"
-futures-util = "0.3.5"
 rand = "0.7"
 redjubjub = "0.2"
+
+metrics = "0.12"
+futures = "0.3.5"
+futures-util = "0.3.5"
 tokio = { version = "0.2.22", features = ["time", "sync", "stream", "tracing"] }
 tower = "0.3"
 tracing = "0.1.17"
 tracing-futures = "0.2.4"
-metrics = "0.12"
 
 tower-batch = { path = "../tower-batch/" }
 zebra-chain = { path = "../zebra-chain" }
 zebra-state = { path = "../zebra-state" }
 
 [dev-dependencies]
-color-eyre = "0.5"
 rand = "0.7"
 spandoc = "0.2"
 tokio = { version = "0.2", features = ["full"] }

zebra-consensus/src/block.rs

@@ -3,7 +3,7 @@
 //! Verification occurs in multiple stages:
 //!   - getting blocks (disk- or network-bound)
 //!   - context-free verification of signatures, proofs, and scripts (CPU-bound)
-//!   - context-dependent verification of the chain state (awaits a verified parent block)
+//!   - context-dependent verification of the chain state (depends on previous blocks)
 //!
 //! Verification is provided via a `tower::Service`, to support backpressure and batch
 //! verification.
@@ -12,6 +12,7 @@
 mod tests;
 
 use chrono::Utc;
+use color_eyre::eyre::{eyre, Report};
 use futures_util::FutureExt;
 use std::{
     error,
@@ -19,10 +20,13 @@ use std::{
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
+    time::Duration,
 };
-use tower::{buffer::Buffer, Service};
+use tokio::time;
+use tower::{buffer::Buffer, Service, ServiceExt};
 
 use zebra_chain::block::{Block, BlockHeaderHash};
+use zebra_chain::types::BlockHeight;
 
 struct BlockVerifier<S>
 where
@@ -67,8 +71,9 @@ where
     }
 
     fn call(&mut self, block: Arc<Block>) -> Self::Future {
-        // TODO(jlusby): Error = Report, handle errors from state_service.
+        let mut state = self.state_service.clone();
 
+        // TODO(jlusby): Error = Report, handle errors from state_service.
         async move {
             // Since errors cause an early exit, try to do the
             // quick checks first.
@@ -78,6 +83,38 @@ where
             block.header.is_equihash_solution_valid()?;
             block.is_coinbase_first()?;
 
+            // These checks only apply to generated blocks. We check the block
+            // height for parsed blocks when we deserialize them.
+            let height = block
+                .coinbase_height()
+                .ok_or("Invalid block: missing block height")?;
+            if height > BlockHeight::MAX {
+                Err("Invalid block height: greater than the maximum height.")?;
+            }
+
+            // As a temporary solution for chain gaps, wait for the previous block,
+            // and check its height.
+            // TODO:
+            //   - Add a previous block height and hash constraint to the AddBlock request,
+            //     so that we can verify in parallel, then check constraints before committing
+            //
+            // Skip contextual checks for the genesis block
+            let previous_block_hash = block.header.previous_block_hash;
+            if previous_block_hash != crate::parameters::GENESIS_PREVIOUS_BLOCK_HASH {
+                tracing::debug!(?height, "Awaiting previous block from state");
+                let previous_block = BlockVerifier::await_block(
+                    &mut state,
+                    previous_block_hash,
+                    BlockHeight(height.0 - 1),
+                )
+                .await?;
+
+                let previous_height = previous_block.coinbase_height().unwrap();
+                if height.0 != previous_height.0 + 1 {
+                    Err("Invalid block height: must be 1 more than the previous block height.")?;
+                }
+            }
+
             // TODO:
             //   - header verification
             //   - contextual verification
@@ -88,6 +125,62 @@ where
     }
 }
 
+/// The BlockVerifier implementation.
+///
+/// The state service is only used for contextual verification.
+/// (The `ChainVerifier` updates the state.)
+impl<S> BlockVerifier<S>
+where
+    S: Service<zebra_state::Request, Response = zebra_state::Response, Error = Error>
+        + Send
+        + Clone
+        + 'static,
+    S::Future: Send + 'static,
+{
+    /// Get the block for `hash`, using `state`.
+    ///
+    /// If there is no block for that hash, returns `Ok(None)`.
+    /// Returns an error if `state.poll_ready` errors.
+    async fn get_block(state: &mut S, hash: BlockHeaderHash) -> Result<Option<Arc<Block>>, Report> {
+        let block = state
+            .ready_and()
+            .await
+            .map_err(|e| eyre!(e))?
+            .call(zebra_state::Request::GetBlock { hash })
+            .await
+            .map(|response| match response {
+                zebra_state::Response::Block { block } => block,
+                _ => unreachable!("GetBlock request can only result in Response::Block"),
+            })
+            .ok();
+
+        Ok(block)
+    }
+
+    /// Wait until a block with `hash` is in `state`.
+    ///
+    /// Returns an error if `state.poll_ready` errors.
+    async fn await_block(
+        state: &mut S,
+        hash: BlockHeaderHash,
+        height: BlockHeight,
+    ) -> Result<Arc<Block>, Report> {
+        loop {
+            match BlockVerifier::get_block(state, hash).await? {
+                Some(block) => return Ok(block),
+                // Busy-waiting is only a temporary solution to waiting for blocks.
+                // TODO:
+                //   - Get an AwaitBlock future from the state
+                //   - Replace with AddBlock constraints
+                None => {
+                    tracing::debug!(?height, ?hash, "Waiting for state to have block");
+                    time::delay_for(Duration::from_secs(2)).await
+                }
+            };
+        }
+    }
+}
+
 /// Return a block verification service, using the provided state service.
 ///
 /// The block verifier holds a state service of type `S`, used as context for

zebra-consensus/src/block/tests.rs

@@ -2,11 +2,8 @@
 
 use super::*;
 
-use chrono::{Duration, Utc};
+use chrono::Utc;
-use color_eyre::eyre::eyre;
+use color_eyre::eyre::{eyre, Report};
-use color_eyre::eyre::Report;
-use std::sync::Arc;
-use tower::{util::ServiceExt, Service};
 
 use zebra_chain::block::Block;
 use zebra_chain::block::BlockHeader;
@@ -23,7 +20,7 @@ async fn verify() -> Result<(), Report> {
     zebra_test::init();
 
     let block =
-        Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
+        Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
     let hash: BlockHeaderHash = block.as_ref().into();
 
     let state_service = Box::new(zebra_state::in_memory::init());
@@ -52,7 +49,7 @@ async fn verify_fail_future_time() -> Result<(), Report> {
     zebra_test::init();
 
     let mut block =
-        <Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
+        <Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
 
     let state_service = zebra_state::in_memory::init();
     let mut block_verifier = super::init(state_service.clone());
@@ -62,7 +59,7 @@ async fn verify_fail_future_time() -> Result<(), Report> {
     // those checks should be performed later in validation, because they
     // are more expensive.
     let three_hours_in_the_future = Utc::now()
-        .checked_add_signed(Duration::hours(3))
+        .checked_add_signed(chrono::Duration::hours(3))
         .ok_or("overflow when calculating 3 hours in the future")
         .map_err(|e| eyre!(e))?;
     block.header.time = three_hours_in_the_future;
@@ -97,7 +94,7 @@ async fn header_solution() -> Result<(), Report> {
     let ready_verifier_service = block_verifier.ready_and().await.map_err(|e| eyre!(e))?;
 
     // Get a valid block
-    let mut block = Block::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])
+    let mut block = Block::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])
         .expect("block test vector should deserialize");
 
     // This should be ok
@@ -166,7 +163,8 @@ async fn coinbase() -> Result<(), Report> {
     let ready_verifier_service = block_verifier.ready_and().await.map_err(|e| eyre!(e))?;
 
     // Test 3: Invalid coinbase position
-    let mut block = Block::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
+    let mut block =
+        Block::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
     assert_eq!(block.transactions.len(), 1);
 
     // Extract the coinbase transaction from the block

zebra-consensus/src/chain/tests.rs

@@ -6,11 +6,11 @@ use crate::checkpoint::CheckpointList;
 
 use color_eyre::eyre::Report;
 use color_eyre::eyre::{bail, eyre};
-use futures::future::TryFutureExt;
+use futures::{future::TryFutureExt, stream::FuturesUnordered};
-use std::mem::drop;
+use std::{collections::BTreeMap, mem::drop, sync::Arc, time::Duration};
-use std::{collections::BTreeMap, sync::Arc, time::Duration};
+use tokio::{stream::StreamExt, time::timeout};
-use tokio::time::timeout;
+use tower::{Service, ServiceExt};
-use tower::{util::ServiceExt, Service};
+use tracing_futures::Instrument;
 
 use zebra_chain::block::{Block, BlockHeader};
 use zebra_chain::serialization::ZcashDeserialize;
@@ -22,6 +22,9 @@ use zebra_chain::Network::{self, *};
 /// If the verifier doesn't send a message on the channel, any tests that
 /// await the channel future will hang.
 ///
+/// The block verifier waits for the previous block to reach the state service.
+/// If that never happens, the test can hang.
+///
 /// This value is set to a large value, to avoid spurious failures due to
 /// high system load.
 const VERIFY_TIMEOUT_SECONDS: u64 = 10;
@@ -126,17 +129,34 @@ async fn verify_block() -> Result<(), Report> {
 
     let (mut chain_verifier, _) = verifiers_from_checkpoint_list(checkpoint_list);
 
+    /// SPANDOC: Make sure the verifier service is ready for block 0
+    let ready_verifier_service = chain_verifier.ready_and().await.map_err(|e| eyre!(e))?;
+    /// SPANDOC: Set up the future for block 0
+    let verify_future = timeout(
+        Duration::from_secs(VERIFY_TIMEOUT_SECONDS),
+        ready_verifier_service.call(block0.clone()),
+    );
+    /// SPANDOC: Verify block 0
+    // TODO(teor || jlusby): check error kind
+    let verify_response = verify_future
+        .map_err(|e| eyre!(e))
+        .await
+        .expect("timeout should not happen")
+        .expect("block should verify");
+
+    assert_eq!(verify_response, hash0);
+
     let block1 = Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_1_BYTES[..])?;
     let hash1: BlockHeaderHash = block1.as_ref().into();
 
-    /// SPANDOC: Make sure the verifier service is ready
+    /// SPANDOC: Make sure the verifier service is ready for block 1
     let ready_verifier_service = chain_verifier.ready_and().await.map_err(|e| eyre!(e))?;
-    /// SPANDOC: Set up the future
+    /// SPANDOC: Set up the future for block 1
     let verify_future = timeout(
         Duration::from_secs(VERIFY_TIMEOUT_SECONDS),
         ready_verifier_service.call(block1.clone()),
     );
-    /// SPANDOC: Verify the block
+    /// SPANDOC: Verify block 1
     // TODO(teor || jlusby): check error kind
     let verify_response = verify_future
         .map_err(|e| eyre!(e))
@@ -371,3 +391,68 @@ async fn verify_fail_add_block_checkpoint() -> Result<(), Report> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn continuous_blockchain_test() -> Result<(), Report> {
+    continuous_blockchain().await
+}
+
+/// Test a continuous blockchain in the BlockVerifier
+#[spandoc::spandoc]
+async fn continuous_blockchain() -> Result<(), Report> {
+    zebra_test::init();
+
+    // A continuous blockchain
+    let mut blockchain = Vec::new();
+    for b in &[
+        &zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_1_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_2_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_3_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_4_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_5_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_6_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_7_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_8_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_9_BYTES[..],
+        &zebra_test::vectors::BLOCK_MAINNET_10_BYTES[..],
+    ] {
+        let block = Arc::<Block>::zcash_deserialize(*b)?;
+        let hash: BlockHeaderHash = block.as_ref().into();
+        blockchain.push((block.clone(), block.coinbase_height().unwrap(), hash));
+    }
+
+    // Make a checkpoint list containing the genesis block
+    let checkpoint_list: BTreeMap<BlockHeight, BlockHeaderHash> = blockchain
+        .iter()
+        .map(|(_, height, hash)| (*height, *hash))
+        .take(1)
+        .collect();
+    let checkpoint_list = CheckpointList::from_list(checkpoint_list).map_err(|e| eyre!(e))?;
+
+    let (mut chain_verifier, _) = verifiers_from_checkpoint_list(checkpoint_list);
+
+    let mut handles = FuturesUnordered::new();
+
+    // Now verify each block
+    for (block, height, _hash) in blockchain {
+        /// SPANDOC: Make sure the verifier service is ready for block {?height}
+        let ready_verifier_service = chain_verifier.ready_and().map_err(|e| eyre!(e)).await?;
+
+        /// SPANDOC: Set up the future for block {?height}
+        let verify_future = timeout(
+            std::time::Duration::from_secs(VERIFY_TIMEOUT_SECONDS),
+            ready_verifier_service.call(block.clone()),
+        );
+
+        /// SPANDOC: spawn verification future in the background for block {?height}
+        let handle = tokio::spawn(verify_future.in_current_span());
+        handles.push(handle);
+    }
+
+    while let Some(result) = handles.next().await {
+        result??.map_err(|e| eyre!(e))?;
+    }
+
+    Ok(())
+}

zebra-consensus/src/checkpoint/tests.rs

@@ -305,7 +305,7 @@ async fn continuous_blockchain(restart_height: Option<BlockHeight>) -> Result<()
                 break;
             }
 
-            /// SPANDOC: Make sure the verifier service is ready
+            /// SPANDOC: Make sure the verifier service is ready for block {?height}
             let ready_verifier_service = checkpoint_verifier
                 .ready_and()
                 .map_err(|e| eyre!(e))
@@ -317,7 +317,7 @@ async fn continuous_blockchain(restart_height: Option<BlockHeight>) -> Result<()
                 ready_verifier_service.call(block.clone()),
             );
 
-            /// SPANDOC: spawn verification future in the background
+            /// SPANDOC: spawn verification future in the background for block {?height}
             let handle = tokio::spawn(verify_future.in_current_span());
             handles.push(handle);