diff --git a/Cargo.lock b/Cargo.lock
index a6a250a4..6277a9e3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2333,6 +2333,9 @@ dependencies = [
  "tokio",
  "tower",
  "tracing",
+ "tracing-error",
+ "tracing-futures",
+ "tracing-subscriber",
  "zebra-chain",
  "zebra-state",
  "zebra-test",
diff --git a/zebra-consensus/Cargo.toml b/zebra-consensus/Cargo.toml
index ac2fbc75..691c2725 100644
--- a/zebra-consensus/Cargo.toml
+++ b/zebra-consensus/Cargo.toml
@@ -19,4 +19,7 @@ zebra-test = { path = "../zebra-test/" }
 spandoc = "0.1"
 tokio = { version = "0.2.21", features = ["full"] }
 tracing = "0.1.15"
+tracing-error = "0.1.2"
+tracing-futures = "0.2"
+tracing-subscriber = "0.2.6"
 color-eyre = "0.5"
diff --git a/zebra-consensus/src/checkpoint.rs b/zebra-consensus/src/checkpoint.rs
index b4a6352e..c3d69e14 100644
--- a/zebra-consensus/src/checkpoint.rs
+++ b/zebra-consensus/src/checkpoint.rs
@@ -155,8 +155,9 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::tests::install_tracing;
 
-    use color_eyre::eyre::{bail, ensure, eyre, Report};
+    use color_eyre::eyre::{bail, eyre, Report};
     use tower::{util::ServiceExt, Service};
 
     use zebra_chain::serialization::ZcashDeserialize;
@@ -164,6 +165,8 @@ mod tests {
     #[tokio::test]
     #[spandoc::spandoc]
     async fn checkpoint_single_item_list() -> Result<(), Report> {
+        install_tracing();
+
         let block0 =
             Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
         let hash0: BlockHeaderHash = block0.as_ref().into();
@@ -178,89 +181,44 @@ mod tests {
         let mut state_service = Box::new(zebra_state::in_memory::init());
         let mut checkpoint_verifier = super::init(state_service.clone(), genesis_checkpoint_list);
 
-        // Verify block 0
-        let verify_response = checkpoint_verifier
+        /// Make sure the verifier service is ready
+        let ready_verifier_service = checkpoint_verifier
             .ready_and()
             .await
-            .map_err(|e| eyre!(e))?
+            .map_err(|e| eyre!(e))?;
+        /// Verify block 0
+        let verify_response = ready_verifier_service
             .call(block0.clone())
             .await
             .map_err(|e| eyre!(e))?;
 
-        ensure!(
-            verify_response == hash0,
-            "unexpected response kind: {:?}",
-            verify_response
-        );
+        assert_eq!(verify_response, hash0);
 
-        let state_response = state_service
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
+        /// Make sure the state service is ready
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Make sure the block was added to the state
+        let state_response = ready_state_service
             .call(zebra_state::Request::GetBlock { hash: hash0 })
             .await
             .map_err(|e| eyre!(e))?;
 
-        match state_response {
-            zebra_state::Response::Block {
-                block: returned_block,
-            } => assert_eq!(block0, returned_block),
-            _ => bail!("unexpected response kind: {:?}", state_response),
+        if let zebra_state::Response::Block {
+            block: returned_block,
+        } = state_response
+        {
+            assert_eq!(block0, returned_block);
+        } else {
+            bail!("unexpected response kind: {:?}", state_response);
         }
 
         Ok(())
     }
 
-    #[tokio::test]
-    #[spandoc::spandoc]
-    async fn checkpoint_list_empty_fail() -> Result<(), Report> {
-        let block0 =
-            Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
-
-        let mut state_service = Box::new(zebra_state::in_memory::init());
-        let mut checkpoint_verifier = super::init(
-            state_service.clone(),
-            <HashMap<BlockHeight, BlockHeaderHash>>::new(),
-        );
-
-        // Try to verify the block, and expect failure
-        let verify_result = checkpoint_verifier
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(block0.clone())
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            verify_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        // Now make sure the block isn't in the state
-        let state_result = state_service
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(zebra_state::Request::GetBlock {
-                hash: block0.as_ref().into(),
-            })
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            state_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        Ok(())
-    }
-
     #[tokio::test]
     #[spandoc::spandoc]
     async fn checkpoint_not_present_fail() -> Result<(), Report> {
+        install_tracing();
+
         let block0 =
             Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
         let block415000 =
@@ -276,53 +234,39 @@ mod tests {
         let mut state_service = Box::new(zebra_state::in_memory::init());
         let mut checkpoint_verifier = super::init(state_service.clone(), genesis_checkpoint_list);
 
-        // Try to verify block 415000, and expect failure
-        let verify_result = checkpoint_verifier
+        /// Make sure the verifier service is ready
+        let ready_verifier_service = checkpoint_verifier
             .ready_and()
             .await
-            .map_err(|e| eyre!(e))?
+            .map_err(|e| eyre!(e))?;
+        /// Try to verify block 415000, and expect failure
+        // TODO(teor || jlusby): check error kind
+        ready_verifier_service
             .call(block415000.clone())
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            verify_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        // Now make sure neither block is in the state
-        let state_result = state_service
-            .ready_and()
             .await
-            .map_err(|e| eyre!(e))?
+            .unwrap_err();
+
+        /// Make sure the state service is ready (1/2)
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Make sure neither block is in the state: expect GetBlock 415000 to fail.
+        // TODO(teor || jlusby): check error kind
+        ready_state_service
             .call(zebra_state::Request::GetBlock {
                 hash: block415000.as_ref().into(),
             })
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            state_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        let state_result = state_service
-            .ready_and()
             .await
-            .map_err(|e| eyre!(e))?
+            .unwrap_err();
+
+        /// Make sure the state service is ready (2/2)
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Make sure neither block is in the state: expect GetBlock 0 to fail.
+        // TODO(teor || jlusby): check error kind
+        ready_state_service
             .call(zebra_state::Request::GetBlock {
                 hash: block0.as_ref().into(),
             })
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            state_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
+            .await
+            .unwrap_err();
 
         Ok(())
     }
@@ -330,6 +274,8 @@ mod tests {
     #[tokio::test]
     #[spandoc::spandoc]
     async fn checkpoint_wrong_hash_fail() -> Result<(), Report> {
+        install_tracing();
+
         let block0 =
             Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
 
@@ -344,37 +290,28 @@ mod tests {
         let mut state_service = Box::new(zebra_state::in_memory::init());
         let mut checkpoint_verifier = super::init(state_service.clone(), genesis_checkpoint_list);
 
-        // Try to verify block 0, and expect failure
-        let verify_result = checkpoint_verifier
+        /// Make sure the verifier service is ready
+        let ready_verifier_service = checkpoint_verifier
             .ready_and()
             .await
-            .map_err(|e| eyre!(e))?
+            .map_err(|e| eyre!(e))?;
+        /// Try to verify block 0, and expect failure
+        // TODO(teor || jlusby): check error kind
+        ready_verifier_service
             .call(block0.clone())
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            verify_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        // Now make sure block 0 is not in the state
-        let state_result = state_service
-            .ready_and()
             .await
-            .map_err(|e| eyre!(e))?
+            .unwrap_err();
+
+        /// Make sure the state service is ready
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Now make sure block 0 is not in the state
+        // TODO(teor || jlusby): check error kind
+        ready_state_service
             .call(zebra_state::Request::GetBlock {
                 hash: block0.as_ref().into(),
             })
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            state_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
+            .await
+            .unwrap_err();
 
         Ok(())
     }
diff --git a/zebra-consensus/src/lib.rs b/zebra-consensus/src/lib.rs
index b0d51372..bf894937 100644
--- a/zebra-consensus/src/lib.rs
+++ b/zebra-consensus/src/lib.rs
@@ -16,3 +16,32 @@
 pub mod checkpoint;
 pub mod mempool;
 pub mod verify;
+
+/// Test utility functions
+///
+/// Submodules have their own specific tests.
+#[cfg(test)]
+mod tests {
+    use std::sync::Once;
+    use tracing_error::ErrorLayer;
+    use tracing_subscriber::prelude::*;
+    use tracing_subscriber::{fmt, EnvFilter};
+
+    static LOGGER_INIT: Once = Once::new();
+
+    // TODO(jlusby): Refactor into the zebra-test crate (#515)
+    pub(crate) fn install_tracing() {
+        LOGGER_INIT.call_once(|| {
+            let fmt_layer = fmt::layer().with_target(false);
+            let filter_layer = EnvFilter::try_from_default_env()
+                .or_else(|_| EnvFilter::try_new("info"))
+                .unwrap();
+
+            tracing_subscriber::registry()
+                .with(filter_layer)
+                .with(fmt_layer)
+                .with(ErrorLayer::default())
+                .init();
+        })
+    }
+}
diff --git a/zebra-consensus/src/verify.rs b/zebra-consensus/src/verify.rs
index e8f983f5..cda5958e 100644
--- a/zebra-consensus/src/verify.rs
+++ b/zebra-consensus/src/verify.rs
@@ -1,333 +1,10 @@
-//! Block verification and chain state updates for Zebra.
+//! Block and transaction verification for Zebra.
 //!
-//! 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)
-//!
-//! Verification is provided via a `tower::Service`, to support backpressure and batch
+//! Verification is provided via `tower::Service`s, to support backpressure and batch
 //! verification.
 
-use chrono::Utc;
-use futures_util::FutureExt;
-use std::{
-    error,
-    future::Future,
-    pin::Pin,
-    sync::Arc,
-    task::{Context, Poll},
-};
-use tower::{buffer::Buffer, Service, ServiceExt};
-
-use zebra_chain::block::{Block, BlockHeaderHash};
-
 mod block;
 mod script;
 mod transaction;
 
-struct BlockVerifier<S> {
-    state_service: S,
-}
-
-/// The error type for the BlockVerifier Service.
-// TODO(jlusby): Error = Report ?
-type Error = Box<dyn error::Error + Send + Sync + 'static>;
-
-/// The BlockVerifier service implementation.
-///
-/// After verification, blocks are added to the underlying state service.
-impl<S> Service<Arc<Block>> for BlockVerifier<S>
-where
-    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, _: &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.
-        // TODO(teor):
-        //   - handle chain reorgs
-        //   - adjust state_service "unique block height" conditions
-        let mut state_service = self.state_service.clone();
-
-        async move {
-            // Since errors cause an early exit, try to do the
-            // quick checks first.
-
-            let now = Utc::now();
-            block::node_time_check(block.header.time, now)?;
-
-            // `Tower::Buffer` requires a 1:1 relationship between `poll()`s
-            // and `call()`s, because it reserves a buffer slot in each
-            // `call()`.
-            let add_block = state_service
-                .ready_and()
-                .await?
-                .call(zebra_state::Request::AddBlock { block });
-
-            match add_block.await? {
-                zebra_state::Response::Added { hash } => Ok(hash),
-                _ => Err("adding block to zebra-state failed".into()),
-            }
-        }
-        .boxed()
-    }
-}
-
-/// Return a block verification service, using the provided state service.
-///
-/// The block verifier holds a state service of type `S`, used as context for
-/// block validation and to which newly verified blocks will be committed. This
-/// state is pluggable to allow for testing or instrumentation.
-///
-/// The returned type is opaque to allow instrumentation or other wrappers, but
-/// can be boxed for storage. It is also `Clone` to allow sharing of a
-/// verification service.
-///
-/// This function should be called only once for a particular state service (and
-/// the result be shared) rather than constructing multiple verification services
-/// backed by the same state layer.
-pub fn init<S>(
-    state_service: S,
-) -> 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(BlockVerifier { state_service }, 1)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    use chrono::{Duration, Utc};
-    use color_eyre::eyre::Report;
-    use color_eyre::eyre::{bail, ensure, eyre};
-    use tower::{util::ServiceExt, Service};
-    use zebra_chain::serialization::ZcashDeserialize;
-
-    #[tokio::test]
-    #[spandoc::spandoc]
-    async fn verify() -> Result<(), Report> {
-        let block =
-            Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
-        let hash: BlockHeaderHash = block.as_ref().into();
-
-        let state_service = Box::new(zebra_state::in_memory::init());
-        let mut block_verifier = super::init(state_service);
-
-        let verify_response = block_verifier
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(block.clone())
-            .await
-            .map_err(|e| eyre!(e))?;
-
-        ensure!(
-            verify_response == hash,
-            "unexpected response kind: {:?}",
-            verify_response
-        );
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    #[spandoc::spandoc]
-    async fn round_trip() -> Result<(), Report> {
-        let block =
-            Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
-        let hash: BlockHeaderHash = block.as_ref().into();
-
-        let mut state_service = zebra_state::in_memory::init();
-        let mut block_verifier = super::init(state_service.clone());
-
-        let verify_response = block_verifier
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(block.clone())
-            .await
-            .map_err(|e| eyre!(e))?;
-
-        ensure!(
-            verify_response == hash,
-            "unexpected response kind: {:?}",
-            verify_response
-        );
-
-        let state_response = state_service
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(zebra_state::Request::GetBlock { hash })
-            .await
-            .map_err(|e| eyre!(e))?;
-
-        match state_response {
-            zebra_state::Response::Block {
-                block: returned_block,
-            } => assert_eq!(block, returned_block),
-            _ => bail!("unexpected response kind: {:?}", state_response),
-        }
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    #[spandoc::spandoc]
-    async fn verify_fail_add_block() -> Result<(), Report> {
-        zebra_test::init();
-
-        let block =
-            Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
-        let hash: BlockHeaderHash = block.as_ref().into();
-
-        let mut state_service = zebra_state::in_memory::init();
-        let mut block_verifier = super::init(state_service.clone());
-
-        // Add the block for the first time
-        let verify_response = block_verifier
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(block.clone())
-            .await
-            .map_err(|e| eyre!(e))?;
-
-        ensure!(
-            verify_response == hash,
-            "unexpected response kind: {:?}",
-            verify_response
-        );
-
-        let state_response = state_service
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(zebra_state::Request::GetBlock { hash })
-            .await
-            .map_err(|e| eyre!(e))?;
-
-        match state_response {
-            zebra_state::Response::Block {
-                block: returned_block,
-            } => assert_eq!(block, returned_block),
-            _ => bail!("unexpected response kind: {:?}", state_response),
-        }
-
-        // Now try to add the block again, verify should fail
-        // TODO(teor): ignore duplicate block verifies?
-        let verify_result = block_verifier
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(block.clone())
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            verify_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        // But the state should still return the original block we added
-        let state_response = state_service
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(zebra_state::Request::GetBlock { hash })
-            .await
-            .map_err(|e| eyre!(e))?;
-
-        match state_response {
-            zebra_state::Response::Block {
-                block: returned_block,
-            } => assert_eq!(block, returned_block),
-            _ => bail!("unexpected response kind: {:?}", state_response),
-        }
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    #[spandoc::spandoc]
-    async fn verify_fail_future_time() -> Result<(), Report> {
-        let mut block =
-            <Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
-
-        let mut state_service = zebra_state::in_memory::init();
-        let mut block_verifier = super::init(state_service.clone());
-
-        // Modify the block's time
-        // Changing the block header also invalidates the header hashes, but
-        // 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))
-            .ok_or("overflow when calculating 3 hours in the future")
-            .map_err(|e| eyre!(e))?;
-        block.header.time = three_hours_in_the_future;
-
-        let arc_block: Arc<Block> = block.into();
-
-        // Try to add the block, and expect failure
-        let verify_result = block_verifier
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(arc_block.clone())
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            verify_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        // Now make sure the block isn't in the state
-        let state_result = state_service
-            .ready_and()
-            .await
-            .map_err(|e| eyre!(e))?
-            .call(zebra_state::Request::GetBlock {
-                hash: arc_block.as_ref().into(),
-            })
-            .await;
-
-        ensure!(
-            // TODO(teor || jlusby): check error string
-            state_result.is_err(),
-            "unexpected result kind: {:?}",
-            verify_result
-        );
-
-        Ok(())
-    }
-}
+pub use block::init;
diff --git a/zebra-consensus/src/verify/block.rs b/zebra-consensus/src/verify/block.rs
index 3a3ec03c..a2791d2a 100644
--- a/zebra-consensus/src/verify/block.rs
+++ b/zebra-consensus/src/verify/block.rs
@@ -1,8 +1,25 @@
-//! Block validity checks
-
-use super::Error;
+//! Block verification and chain state updates for Zebra.
+//!
+//! 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)
+//!
+//! Verification is provided via a `tower::Service`, to support backpressure and batch
+//! verification.
 
 use chrono::{DateTime, Duration, Utc};
+use futures_util::FutureExt;
+use std::{
+    error,
+    future::Future,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tower::{buffer::Buffer, Service, ServiceExt};
+
+use zebra_chain::block::{Block, BlockHeaderHash};
 
 /// Check if `block_header_time` is less than or equal to
 /// 2 hours in the future, according to the node's local clock (`now`).
@@ -18,7 +35,7 @@ use chrono::{DateTime, Duration, Utc};
 /// accepted."[S 7.5][7.5]
 ///
 /// [7.5]: https://zips.z.cash/protocol/protocol.pdf#blockheader
-pub(super) fn node_time_check(
+pub(crate) fn node_time_check(
     block_header_time: DateTime<Utc>,
     now: DateTime<Utc>,
 ) -> Result<(), Error> {
@@ -33,11 +50,112 @@ pub(super) fn node_time_check(
     }
 }
 
+struct BlockVerifier<S> {
+    /// The underlying `ZebraState`, possibly wrapped in other services.
+    state_service: S,
+}
+
+/// The error type for the BlockVerifier Service.
+// TODO(jlusby): Error = Report ?
+type Error = Box<dyn error::Error + Send + Sync + 'static>;
+
+/// The BlockVerifier service implementation.
+///
+/// After verification, blocks are added to the underlying state service.
+impl<S> Service<Arc<Block>> for BlockVerifier<S>
+where
+    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, _: &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.
+        // TODO(teor):
+        //   - handle chain reorgs
+        //   - adjust state_service "unique block height" conditions
+        let mut state_service = self.state_service.clone();
+
+        async move {
+            // Since errors cause an early exit, try to do the
+            // quick checks first.
+
+            let now = Utc::now();
+            node_time_check(block.header.time, now)?;
+
+            // `Tower::Buffer` requires a 1:1 relationship between `poll()`s
+            // and `call()`s, because it reserves a buffer slot in each
+            // `call()`.
+            let add_block = state_service
+                .ready_and()
+                .await?
+                .call(zebra_state::Request::AddBlock { block });
+
+            match add_block.await? {
+                zebra_state::Response::Added { hash } => Ok(hash),
+                _ => Err("adding block to zebra-state failed".into()),
+            }
+        }
+        .boxed()
+    }
+}
+
+/// Return a block verification service, using the provided state service.
+///
+/// The block verifier holds a state service of type `S`, used as context for
+/// block validation and to which newly verified blocks will be committed. This
+/// state is pluggable to allow for testing or instrumentation.
+///
+/// The returned type is opaque to allow instrumentation or other wrappers, but
+/// can be boxed for storage. It is also `Clone` to allow sharing of a
+/// verification service.
+///
+/// This function should be called only once for a particular state service (and
+/// the result be shared) rather than constructing multiple verification services
+/// backed by the same state layer.
+pub fn init<S>(
+    state_service: S,
+) -> 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(BlockVerifier { state_service }, 1)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::tests::install_tracing;
+
     use chrono::offset::{LocalResult, TimeZone};
+    use chrono::{Duration, Utc};
+    use color_eyre::eyre::Report;
+    use color_eyre::eyre::{bail, eyre};
     use std::sync::Arc;
+    use tower::{util::ServiceExt, Service};
 
     use zebra_chain::block::Block;
     use zebra_chain::serialization::ZcashDeserialize;
@@ -171,4 +289,186 @@ mod tests {
                 .expect("the inverse comparison should be valid");
         }
     }
+
+    #[tokio::test]
+    #[spandoc::spandoc]
+    async fn verify() -> Result<(), Report> {
+        install_tracing();
+
+        let block =
+            Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
+        let hash: BlockHeaderHash = block.as_ref().into();
+
+        let state_service = Box::new(zebra_state::in_memory::init());
+        let mut block_verifier = super::init(state_service);
+
+        /// Make sure the verifier service is ready
+        let ready_verifier_service = block_verifier.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Verify the block
+        let verify_response = ready_verifier_service
+            .call(block.clone())
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        assert_eq!(verify_response, hash);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    #[spandoc::spandoc]
+    async fn round_trip() -> Result<(), Report> {
+        install_tracing();
+
+        let block =
+            Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
+        let hash: BlockHeaderHash = block.as_ref().into();
+
+        let mut state_service = zebra_state::in_memory::init();
+        let mut block_verifier = super::init(state_service.clone());
+
+        /// Make sure the verifier service is ready
+        let ready_verifier_service = block_verifier.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Verify the block
+        let verify_response = ready_verifier_service
+            .call(block.clone())
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        assert_eq!(verify_response, hash);
+
+        /// Make sure the state service is ready
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Make sure the block was added to the state
+        let state_response = ready_state_service
+            .call(zebra_state::Request::GetBlock { hash })
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        if let zebra_state::Response::Block {
+            block: returned_block,
+        } = state_response
+        {
+            assert_eq!(block, returned_block);
+        } else {
+            bail!("unexpected response kind: {:?}", state_response);
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    #[spandoc::spandoc]
+    async fn verify_fail_add_block() -> Result<(), Report> {
+        install_tracing();
+
+        let block =
+            Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
+        let hash: BlockHeaderHash = block.as_ref().into();
+
+        let mut state_service = zebra_state::in_memory::init();
+        let mut block_verifier = super::init(state_service.clone());
+
+        /// Make sure the verifier service is ready (1/2)
+        let ready_verifier_service = block_verifier.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Verify the block for the first time
+        let verify_response = ready_verifier_service
+            .call(block.clone())
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        assert_eq!(verify_response, hash);
+
+        /// Make sure the state service is ready (1/2)
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Make sure the block was added to the state
+        let state_response = ready_state_service
+            .call(zebra_state::Request::GetBlock { hash })
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        if let zebra_state::Response::Block {
+            block: returned_block,
+        } = state_response
+        {
+            assert_eq!(block, returned_block);
+        } else {
+            bail!("unexpected response kind: {:?}", state_response);
+        }
+
+        /// Make sure the verifier service is ready (2/2)
+        let ready_verifier_service = block_verifier.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Now try to add the block again, verify should fail
+        // TODO(teor): ignore duplicate block verifies?
+        // TODO(teor || jlusby): check error kind
+        ready_verifier_service
+            .call(block.clone())
+            .await
+            .unwrap_err();
+
+        /// Make sure the state service is ready (2/2)
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// But the state should still return the original block we added
+        let state_response = ready_state_service
+            .call(zebra_state::Request::GetBlock { hash })
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        if let zebra_state::Response::Block {
+            block: returned_block,
+        } = state_response
+        {
+            assert_eq!(block, returned_block);
+        } else {
+            bail!("unexpected response kind: {:?}", state_response);
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    #[spandoc::spandoc]
+    async fn verify_fail_future_time() -> Result<(), Report> {
+        install_tracing();
+
+        let mut block =
+            <Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_415000_BYTES[..])?;
+
+        let mut state_service = zebra_state::in_memory::init();
+        let mut block_verifier = super::init(state_service.clone());
+
+        // Modify the block's time
+        // Changing the block header also invalidates the header hashes, but
+        // 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))
+            .ok_or("overflow when calculating 3 hours in the future")
+            .map_err(|e| eyre!(e))?;
+        block.header.time = three_hours_in_the_future;
+
+        let arc_block: Arc<Block> = block.into();
+
+        /// Make sure the verifier service is ready
+        let ready_verifier_service = block_verifier.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Try to add the block, and expect failure
+        // TODO(teor || jlusby): check error kind
+        ready_verifier_service
+            .call(arc_block.clone())
+            .await
+            .unwrap_err();
+
+        /// Make sure the state service is ready (2/2)
+        let ready_state_service = state_service.ready_and().await.map_err(|e| eyre!(e))?;
+        /// Now make sure the block isn't in the state
+        // TODO(teor || jlusby): check error kind
+        ready_state_service
+            .call(zebra_state::Request::GetBlock {
+                hash: arc_block.as_ref().into(),
+            })
+            .await
+            .unwrap_err();
+
+        Ok(())
+    }
 }