state: don't pre-buffer the service

There's no reason to return a pre-Buffer'd service (there's no need for
internal access to the state service, as in zebra-network), but wrapping
it internally removes control of the buffer size from the caller.

zebra-consensus/src/block/tests.rs

@@ -114,7 +114,10 @@ async fn check_transcripts() -> Result<(), Report> {
     zebra_test::init();
 
     let network = Network::Mainnet;
-    let state_service = zebra_state::init(zebra_state::Config::ephemeral(), network);
+    let state_service = Buffer::new(
+        zebra_state::init(zebra_state::Config::ephemeral(), network),
+        1,
+    );
 
     let block_verifier = Buffer::new(BlockVerifier::new(network, state_service.clone()), 1);
 

zebra-consensus/src/chain/tests.rs

@@ -4,7 +4,7 @@ use std::{sync::Arc, time::Duration};
 
 use color_eyre::eyre::Report;
 use once_cell::sync::Lazy;
-use tower::{layer::Layer, timeout::TimeoutLayer, Service};
+use tower::{layer::Layer, timeout::TimeoutLayer, Service, ServiceBuilder};
 
 use zebra_chain::{
     block::{self, Block},
@@ -63,7 +63,9 @@ async fn verifiers_from_network(
         + Clone
         + 'static,
 ) {
-    let state_service = zs::init(zs::Config::ephemeral(), network);
+    let state_service = ServiceBuilder::new()
+        .buffer(1)
+        .service(zs::init(zs::Config::ephemeral(), network));
     let chain_verifier =
         crate::chain::init(Config::default(), network, state_service.clone()).await;
 
@@ -152,7 +154,9 @@ async fn verify_checkpoint(config: Config) -> Result<(), Report> {
     let chain_verifier = super::init(
         config.clone(),
         network,
-        zs::init(zs::Config::ephemeral(), network),
+        ServiceBuilder::new()
+            .buffer(1)
+            .service(zs::init(zs::Config::ephemeral(), network)),
     )
     .await;
 

zebra-consensus/src/checkpoint/tests.rs

@@ -9,7 +9,7 @@ use color_eyre::eyre::{eyre, Report};
 use futures::{future::TryFutureExt, stream::FuturesUnordered};
 use std::{cmp::min, mem::drop, time::Duration};
 use tokio::{stream::StreamExt, time::timeout};
-use tower::{Service, ServiceExt};
+use tower::{Service, ServiceBuilder, ServiceExt};
 use tracing_futures::Instrument;
 
 use zebra_chain::parameters::Network::*;
@@ -45,7 +45,9 @@ async fn single_item_checkpoint_list() -> Result<(), Report> {
             .cloned()
             .collect();
 
-    let state_service = zebra_state::init(zebra_state::Config::ephemeral(), Mainnet);
+    let state_service = ServiceBuilder::new()
+        .buffer(1)
+        .service(zebra_state::init(zebra_state::Config::ephemeral(), Mainnet));
     let mut checkpoint_verifier =
         CheckpointVerifier::from_list(genesis_checkpoint_list, None, state_service)
             .map_err(|e| eyre!(e))?;
@@ -127,7 +129,9 @@ async fn multi_item_checkpoint_list() -> Result<(), Report> {
         .map(|(_block, height, hash)| (*height, *hash))
         .collect();
 
-    let state_service = zebra_state::init(zebra_state::Config::ephemeral(), Mainnet);
+    let state_service = ServiceBuilder::new()
+        .buffer(1)
+        .service(zebra_state::init(zebra_state::Config::ephemeral(), Mainnet));
     let mut checkpoint_verifier =
         CheckpointVerifier::from_list(checkpoint_list, None, state_service)
             .map_err(|e| eyre!(e))?;
@@ -268,7 +272,7 @@ async fn continuous_blockchain(restart_height: Option<block::Height>) -> Result<
         let initial_tip = restart_height.map(|block::Height(height)| {
             (blockchain[height as usize].1, blockchain[height as usize].2)
         });
-        let state_service = zebra_state::init(zebra_state::Config::ephemeral(), Mainnet);
+        let state_service = ServiceBuilder::new().buffer(1).service(zebra_state::init(zebra_state::Config::ephemeral(), Mainnet));
         let mut checkpoint_verifier =
             CheckpointVerifier::from_list(checkpoint_list, initial_tip, state_service.clone())
                 .map_err(|e| eyre!(e))?;
@@ -402,7 +406,9 @@ async fn block_higher_than_max_checkpoint_fail() -> Result<(), Report> {
             .cloned()
             .collect();
 
-    let state_service = zebra_state::init(zebra_state::Config::ephemeral(), Mainnet);
+    let state_service = ServiceBuilder::new()
+        .buffer(1)
+        .service(zebra_state::init(zebra_state::Config::ephemeral(), Mainnet));
     let mut checkpoint_verifier =
         CheckpointVerifier::from_list(genesis_checkpoint_list, None, state_service)
             .map_err(|e| eyre!(e))?;
@@ -479,7 +485,9 @@ async fn wrong_checkpoint_hash_fail() -> Result<(), Report> {
             .cloned()
             .collect();
 
-    let state_service = zebra_state::init(zebra_state::Config::ephemeral(), Mainnet);
+    let state_service = ServiceBuilder::new()
+        .buffer(1)
+        .service(zebra_state::init(zebra_state::Config::ephemeral(), Mainnet));
     let mut checkpoint_verifier =
         CheckpointVerifier::from_list(genesis_checkpoint_list, None, state_service)
             .map_err(|e| eyre!(e))?;
@@ -661,7 +669,9 @@ async fn checkpoint_drop_cancel() -> Result<(), Report> {
         .map(|(_block, height, hash)| (*height, *hash))
         .collect();
 
-    let state_service = zebra_state::init(zebra_state::Config::ephemeral(), Mainnet);
+    let state_service = ServiceBuilder::new()
+        .buffer(1)
+        .service(zebra_state::init(zebra_state::Config::ephemeral(), Mainnet));
     let mut checkpoint_verifier =
         CheckpointVerifier::from_list(checkpoint_list, None, state_service)
             .map_err(|e| eyre!(e))?;
@@ -747,7 +757,9 @@ async fn hard_coded_mainnet() -> Result<(), Report> {
         Arc::<Block>::zcash_deserialize(&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..])?;
     let hash0 = block0.hash();
 
-    let state_service = zebra_state::init(zebra_state::Config::ephemeral(), Mainnet);
+    let state_service = ServiceBuilder::new()
+        .buffer(1)
+        .service(zebra_state::init(zebra_state::Config::ephemeral(), Mainnet));
     // Use the hard-coded checkpoint list
     let mut checkpoint_verifier = CheckpointVerifier::new(Network::Mainnet, None, state_service);
 

zebra-state/src/service.rs

@@ -10,7 +10,7 @@ use std::{
 use futures::future::{FutureExt, TryFutureExt};
 use memory_state::{NonFinalizedState, QueuedBlocks};
 use tokio::sync::broadcast;
-use tower::{buffer::Buffer, util::BoxService, Service};
+use tower::{util::BoxService, Service};
 use tracing::instrument;
 use zebra_chain::{
     block::{self, Block},
@@ -253,13 +253,10 @@ impl Service<Request> for StateService {
 ///
 /// Each `network` has its own separate sled database.
 ///
-/// The resulting service is clonable, to provide shared access to a common chain
-/// state. It's possible to construct multiple state services in the same
-/// application (as long as they, e.g., use different storage locations), but
-/// doing so is probably not what you want.
-pub fn init(
-    config: Config,
-    network: Network,
-) -> Buffer<BoxService<Request, Response, BoxError>, Request> {
-    Buffer::new(BoxService::new(StateService::new(config, network)), 3)
+/// To share access to the state, wrap the returned service in a `Buffer`. It's
+/// possible to construct multiple state services in the same application (as
+/// long as they, e.g., use different storage locations), but doing so is
+/// probably not what you want.
+pub fn init(config: Config, network: Network) -> BoxService<Request, Response, BoxError> {
+    BoxService::new(StateService::new(config, network))
 }

zebrad/src/commands/start.rs

@@ -45,7 +45,10 @@ impl StartCmd {
         info!(?config);
 
         info!("initializing node state");
-        let state = zebra_state::init(config.state.clone(), config.network.network);
+        let state = ServiceBuilder::new().buffer(20).service(zebra_state::init(
+            config.state.clone(),
+            config.network.network,
+        ));
 
         info!("initializing chain verifier");
         let verifier = zebra_consensus::chain::init(