consensus: Add a checkpoint verifier stub

zebra-consensus/src/checkpoint.rs

@@ -20,7 +20,7 @@ use std::{
     sync::Arc,
     task::{Context, Poll},
 };
-use tower::{buffer::Buffer, Service};
+use tower::{Service, ServiceExt};
 
 use zebra_chain::block::{Block, BlockHeaderHash};
 use zebra_chain::types::BlockHeight;
@@ -35,7 +35,7 @@ struct CheckpointVerifier<S> {
     /// which only happen in the last few hundred blocks in the chain.
     /// (zcashd allows chain reorganizations up to 99 blocks, and prunes
     /// orphaned side-chains after 288 blocks.)
-    checkpoint_list: Option<HashMap<BlockHeight, BlockHeaderHash>>,
+    checkpoint_list: Arc<HashMap<BlockHeight, BlockHeaderHash>>,
 }
 
 /// The error type for the CheckpointVerifier Service.
@@ -47,63 +47,64 @@ type Error = Box<dyn error::Error + Send + Sync + 'static>;
 /// After verification, blocks are added to the underlying state service.
 impl<S> Service<Arc<Block>> for CheckpointVerifier<S>
 where
-    S: Service<zebra_state::Request, Response = zebra_state::Response, Error = Error>,
+    S: Service<zebra_state::Request, Response = zebra_state::Response, Error = Error>
+        + Send
+        + Clone
+        + 'static,
     S::Future: Send + 'static,
 {
     type Response = BlockHeaderHash;
     type Error = Error;
     type Future =
         Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 
-    fn poll_ready(&mut self, context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.state_service.poll_ready(context)
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        // We don't expect the state to exert backpressure on verifier users,
+        // so we don't need to call `state_service.poll_ready()` here.
+        Poll::Ready(Ok(()))
     }
 
     fn call(&mut self, block: Arc<Block>) -> Self::Future {
         // TODO(jlusby): Error = Report, handle errors from state_service.
-
-        // These checks are cheap, so we can do them in the call()
-
-        let checkpoints = match &self.checkpoint_list {
-            Some(checkpoints) => checkpoints,
-            None => return async { Err("the checkpoint list is empty".into()) }.boxed(),
-        };
-
-        let block_height = match block.coinbase_height() {
-            Some(height) => height,
-            None => {
-                return async { Err("the block does not have a coinbase height".into()) }.boxed()
-            }
-        };
-
-        // TODO(teor):
-        //   - implement chaining from checkpoints to their ancestors
-        //   - if chaining is expensive, move this check to the Future
-        //   - should the state contain a mapping from previous_block_hash to block?
-        let checkpoint_hash_ref = match checkpoints.get(&block_height) {
-            Some(hash) => hash,
-            None => {
-                return async { Err("the block's height is not a checkpoint height".into()) }
-                    .boxed()
-            }
-        };
-
-        // Avoid moving a reference into the future.
-        let checkpoint_hash = *checkpoint_hash_ref;
-
-        // `state_service.call` is OK here because we already called
-        // `state_service.poll_ready` in our `poll_ready`.
-        let add_block = self.state_service.call(zebra_state::Request::AddBlock {
-            block: block.clone(),
-        });
+        let mut state_service = self.state_service.clone();
+        let checkpoint_list = self.checkpoint_list.clone();
 
         async move {
-            // Hashing is expensive, so we do it in the Future
+            if checkpoint_list.is_empty() {
+                return Err("the checkpoint list is empty".into());
+            };
+
+            let block_height = match block.coinbase_height() {
+                Some(height) => height,
+                None => return Err("the block does not have a coinbase height".into()),
+            };
+
+            // TODO(teor):
+            //   - implement chaining from checkpoints to their ancestors
+            //   - if chaining is expensive, move this check to the Future
+            //   - should the state contain a mapping from previous_block_hash to block?
+            let checkpoint_hash = match checkpoint_list.get(&block_height) {
+                Some(&hash) => hash,
+                None => return Err("the block's height is not a checkpoint height".into()),
+            };
+
+            // Hashing is expensive, so we do it as late as possible
             if BlockHeaderHash::from(block.as_ref()) != checkpoint_hash {
                 // The block is on a side-chain
                 return Err("the block hash does not match the checkpoint hash".into());
             }
 
+            // `Tower::Buffer` requires a 1:1 relationship between `poll()`s
+            // and `call()`s, because it reserves a buffer slot in each
+            // `call()`.
+            // TODO(teor): what happens if the await fails?
+            let add_block = state_service
+                .ready_and()
+                .await?
+                .call(zebra_state::Request::AddBlock {
+                    block: block.clone(),
+                });
+
             match add_block.await? {
                 zebra_state::Response::Added { hash } => Ok(hash),
                 _ => Err("adding block to zebra-state failed".into()),
@@ -113,8 +114,10 @@ where
     }
 }
 
-// TODO(teor): add a function for the maximum checkpoint height
-// We can pre-calculate the result in init(), if we want.
+// TODO(teor):
+//   - add a function for the maximum checkpoint height
+//     (We can pre-calculate the result in init(), if we want.)
+//   - check that block.coinbase_height() <= max_checkpoint_height
 
 /// Return a checkpoint verification service, using the provided state service.
 ///
@@ -131,28 +134,25 @@ where
 /// backed by the same state layer.
 pub fn init<S>(
     state_service: S,
-    checkpoint_list: Option<HashMap<BlockHeight, BlockHeaderHash>>,
+    checkpoint_list: impl Into<Arc<HashMap<BlockHeight, BlockHeaderHash>>>,
 ) -> impl Service<
     Arc<Block>,
     Response = BlockHeaderHash,
     Error = Error,
     Future = impl Future<Output = Result<BlockHeaderHash, Error>>,
 > + Send
-       + Clone
        + 'static
 where
     S: Service<zebra_state::Request, Response = zebra_state::Response, Error = Error>
         + Send
+        + Clone
         + 'static,
     S::Future: Send + 'static,
 {
-    Buffer::new(
-        CheckpointVerifier {
-            state_service,
-            checkpoint_list,
-        },
-        1,
-    )
+    CheckpointVerifier {
+        state_service,
+        checkpoint_list: checkpoint_list.into(),
+    }
 }
 
 // TODO(teor): tests