Add initial version of zebra-state (#414)

* rename zebra-storage to zebra-state

* Setup initial skeleton for zebra-state

* add test

* Apply suggestions from code review

Co-authored-by: Henry de Valence <hdevalence@hdevalence.ca>

* move shared test vectors to a common crate

Co-authored-by: Jane Lusby <jane@zfnd.org>
Co-authored-by: Henry de Valence <hdevalence@hdevalence.ca>

Cargo.lock

@@ -975,12 +975,24 @@ dependencies = [
  "kernel32-sys",
  "libc",
  "log",
- "miow",
+ "miow 0.2.1",
  "net2",
  "slab",
  "winapi 0.2.8",
 ]
 
+[[package]]
+name = "mio-named-pipes"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f5e374eff525ce1c5b7687c4cef63943e7686524a387933ad27ca7ec43779cb3"
+dependencies = [
+ "log",
+ "mio",
+ "miow 0.3.4",
+ "winapi 0.3.8",
+]
+
 [[package]]
 name = "mio-uds"
 version = "0.6.7"
@@ -1004,6 +1016,16 @@ dependencies = [
  "ws2_32-sys",
 ]
 
+[[package]]
+name = "miow"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22dfdd1d51b2639a5abd17ed07005c3af05fb7a2a3b1a1d0d7af1000a520c1c7"
+dependencies = [
+ "socket2",
+ "winapi 0.3.8",
+]
+
 [[package]]
 name = "net2"
 version = "0.2.33"
@@ -1680,10 +1702,25 @@ dependencies = [
  "libc",
  "memchr",
  "mio",
+ "mio-named-pipes",
  "mio-uds",
  "num_cpus",
  "pin-project-lite",
+ "signal-hook-registry",
  "slab",
+ "tokio-macros",
+ "winapi 0.3.8",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389"
+dependencies = [
+ "proc-macro2 1.0.9",
+ "quote 1.0.3",
+ "syn 1.0.17",
 ]
 
 [[package]]
@@ -2077,6 +2114,7 @@ dependencies = [
  "sha2",
  "thiserror",
  "x25519-dalek",
+ "zebra-test-vectors",
 ]
 
 [[package]]
@@ -2123,8 +2161,27 @@ name = "zebra-script"
 version = "0.1.0"
 
 [[package]]
-name = "zebra-storage"
+name = "zebra-state"
 version = "0.1.0"
+dependencies = [
+ "color-eyre",
+ "eyre",
+ "futures",
+ "hex",
+ "lazy_static",
+ "tokio",
+ "tower",
+ "zebra-chain",
+ "zebra-test-vectors",
+]
+
+[[package]]
+name = "zebra-test-vectors"
+version = "0.1.0"
+dependencies = [
+ "hex",
+ "lazy_static",
+]
 
 [[package]]
 name = "zebrad"
@@ -2151,6 +2208,7 @@ dependencies = [
  "tracing-log",
  "zebra-chain",
  "zebra-network",
+ "zebra-state",
 ]
 
 [[package]]

Cargo.toml

@@ -2,11 +2,12 @@
 members = [
         "zebra-chain",
         "zebra-network",
-        "zebra-storage",
+        "zebra-state",
         "zebra-script",
         "zebra-consensus",
         "zebra-rpc",
         "zebra-client",
+        "zebra-test-vectors",
         "zebrad",
 ]
 

design/design.md

@@ -21,15 +21,15 @@ The following are general desiderata for Zebra:
   is usually a proxy for this desideratum, but is not exactly the same:
   for instance, a collection of crates like the tokio crates are all
   developed together and have one trust boundary.
-  
+
 * Zebra should be well-factored internally into a collection of
   component libraries which can be used by other applications to
   perform Zcash-related tasks.  Implementation details of each
   component should not leak into all other components.
-  
+
 * Zebra should checkpoint on Sapling activation and drop all
   Sprout-related functionality not required post-Sapling.
-  
+
 Internal Structure
 ==================
 
@@ -98,7 +98,7 @@ All peerset management (finding new peers, creating new outbound
 connections, etc) is completely encapsulated, as is responsibility for
 routing outbound requests to appropriate peers.
 
-`zebra-storage`
+`zebra-state`
 ----------------
 
 ### Internal Dependencies
@@ -157,7 +157,7 @@ for Zcash script inspection, debugging, etc.
 ### Internal Dependencies
 
 - `zebra-chain`
-- `zebra-storage`
+- `zebra-state`
 - `zebra-script`
 
 ### Responsible for
@@ -194,7 +194,7 @@ for Zcash script inspection, debugging, etc.
 ### Internal Dependencies
 
 - `zebra-chain` for structure definitions
-- `zebra-storage` for transaction queries and client/wallet state storage
+- `zebra-state` for transaction queries and client/wallet state storage
 - `zebra-script` possibly? for constructing transactions
 
 ### Responsible for
@@ -203,7 +203,7 @@ for Zcash script inspection, debugging, etc.
 - would be used to implement a wallet
 - create transactions, monitors shielded wallet state, etc.
 
-### Notes 
+### Notes
 
 Communication between the client code and the rest of the node should be done
 by a tower service interface. Since the `Service` trait can abstract from a
@@ -229,7 +229,7 @@ and connects them to each other.
 
 - `zebra-chain`
 - `zebra-network`
-- `zebra-storage`
+- `zebra-state`
 - `zebra-consensus`
 - `zebra-client`
 - `zebra-rpc`

zebra-chain/Cargo.toml

@@ -32,3 +32,4 @@ redjubjub = "0.1"
 [dev-dependencies]
 proptest = "0.10"
 proptest-derive = "0.2.0"
+zebra-test-vectors = { path = "../zebra-test-vectors/" }

zebra-chain/src/block.rs

@@ -1,8 +1,6 @@
 //! Definitions of block datastructures.
 #![allow(clippy::unit_arg)]
 
-#[cfg(test)]
-pub mod test_vectors;
 #[cfg(test)]
 mod tests;
 

zebra-chain/src/block/tests.rs

@@ -104,22 +104,23 @@ fn blockheaderhash_from_blockheader() {
 #[test]
 fn deserialize_blockheader() {
     // https://explorer.zcha.in/blocks/415000
-    let _header = BlockHeader::zcash_deserialize(&test_vectors::HEADER_MAINNET_415000_BYTES[..])
+    let _header =
-        .expect("blockheader test vector should deserialize");
+        BlockHeader::zcash_deserialize(&zebra_test_vectors::HEADER_MAINNET_415000_BYTES[..])
+            .expect("blockheader test vector should deserialize");
 }
 
 #[test]
 fn deserialize_block() {
-    Block::zcash_deserialize(&test_vectors::BLOCK_MAINNET_GENESIS_BYTES[..])
+    Block::zcash_deserialize(&zebra_test_vectors::BLOCK_MAINNET_GENESIS_BYTES[..])
         .expect("block test vector should deserialize");
-    Block::zcash_deserialize(&test_vectors::BLOCK_MAINNET_1_BYTES[..])
+    Block::zcash_deserialize(&zebra_test_vectors::BLOCK_MAINNET_1_BYTES[..])
         .expect("block test vector should deserialize");
     // https://explorer.zcha.in/blocks/415000
-    Block::zcash_deserialize(&test_vectors::BLOCK_MAINNET_415000_BYTES[..])
+    Block::zcash_deserialize(&zebra_test_vectors::BLOCK_MAINNET_415000_BYTES[..])
         .expect("block test vector should deserialize");
     // https://explorer.zcha.in/blocks/434873
     // this one has a bad version field
-    Block::zcash_deserialize(&test_vectors::BLOCK_MAINNET_434873_BYTES[..])
+    Block::zcash_deserialize(&zebra_test_vectors::BLOCK_MAINNET_434873_BYTES[..])
         .expect("block test vector should deserialize");
 }
 

zebra-state/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "zebra-state"
+version = "0.1.0"
+authors = ["Zcash Foundation <zebra@zfnd.org>"]
+license = "MIT OR Apache-2.0"
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+zebra-chain = { path = "../zebra-chain" }
+tower = "0.3.1"
+eyre = "0.4.2"
+futures = "0.3.5"
+lazy_static = "1.4.0"
+hex = "0.4.2"
+
+[dev-dependencies]
+color-eyre = "0.3.2"
+eyre = "0.4.2"
+tokio = { version = "0.2.21", features = ["full"] }
+zebra-test-vectors = { path = "../zebra-test-vectors/" }

zebra-state/src/lib.rs

@@ -0,0 +1,123 @@
+#![doc(html_logo_url = "https://www.zfnd.org/images/zebra-icon.png")]
+#![doc(html_root_url = "https://doc.zebra.zfnd.org/zebra_state")]
+use futures::prelude::*;
+use std::{
+    collections::HashMap,
+    error::Error,
+    future::Future,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tower::{buffer::Buffer, Service};
+use zebra_chain::block::{Block, BlockHeaderHash};
+
+#[derive(Debug)]
+pub enum Request {
+    // TODO(jlusby): deprecate in the future based on our validation story
+    AddBlock { block: Arc<Block> },
+    GetBlock { hash: BlockHeaderHash },
+}
+
+#[derive(Debug)]
+pub enum Response {
+    Added,
+    Block { block: Arc<Block> },
+}
+
+pub mod in_memory {
+    use super::*;
+    use std::error::Error;
+
+    pub fn init() -> impl Service<
+        Request,
+        Response = Response,
+        Error = Box<dyn Error + Send + Sync + 'static>,
+        Future = impl Future<Output = Result<Response, Box<dyn Error + Send + Sync + 'static>>>,
+    > + Send
+           + Clone
+           + 'static {
+        Buffer::new(ZebraState::default(), 1)
+    }
+}
+
+#[derive(Default)]
+struct ZebraState {
+    blocks: HashMap<BlockHeaderHash, Arc<Block>>,
+}
+
+impl Service<Request> for ZebraState {
+    type Response = Response;
+    type Error = Box<dyn Error + Send + Sync + 'static>;
+    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>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: Request) -> Self::Future {
+        match req {
+            Request::AddBlock { block } => {
+                let hash = block.as_ref().into();
+                self.blocks.insert(hash, block);
+
+                async { Ok(Response::Added) }.boxed()
+            }
+            Request::GetBlock { hash } => {
+                let result = self
+                    .blocks
+                    .get(&hash)
+                    .cloned()
+                    .map(|block| Response::Block { block })
+                    .ok_or_else(|| "block could not be found".into());
+
+                async move { result }.boxed()
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use color_eyre::Report;
+    use eyre::{bail, ensure, eyre};
+    use zebra_chain::serialization::ZcashDeserialize;
+
+    #[tokio::test]
+    async fn round_trip() -> Result<(), Report> {
+        let block: Arc<_> =
+            Block::zcash_deserialize(&zebra_test_vectors::BLOCK_MAINNET_415000_BYTES[..])?.into();
+        let hash = block.as_ref().into();
+
+        let mut service = in_memory::init();
+
+        let response = service
+            .call(Request::AddBlock {
+                block: block.clone(),
+            })
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        ensure!(
+            matches!(response, Response::Added),
+            "unexpected response kind: {:?}",
+            response
+        );
+
+        let block_response = service
+            .call(Request::GetBlock { hash })
+            .await
+            .map_err(|e| eyre!(e))?;
+
+        match block_response {
+            Response::Block {
+                block: returned_block,
+            } => assert_eq!(block, returned_block),
+            _ => bail!("unexpected response kind: {:?}", block_response),
+        }
+
+        Ok(())
+    }
+}

zebra-storage/src/lib.rs

@@ -1,10 +0,0 @@
-#![doc(html_logo_url = "https://www.zfnd.org/images/zebra-icon.png")]
-#![doc(html_root_url = "https://doc.zebra.zfnd.org/zebra_storage")]
-
-#[cfg(test)]
-mod tests {
-    #[test]
-    fn it_works() {
-        assert_eq!(2 + 2, 4);
-    }
-}

zebra-test-vectors/Cargo.toml

@@ -1,10 +1,11 @@
 [package]
-name = "zebra-storage"
+name = "zebra-test-vectors"
 version = "0.1.0"
-authors = ["Zcash Foundation <zebra@zfnd.org>"]
+authors = ["Jane Lusby <jane@zfnd.org>"]
-license = "MIT OR Apache-2.0"
 edition = "2018"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+hex = "0.4.2"
+lazy_static = "1.4.0"

zebrad/Cargo.toml

@@ -32,6 +32,7 @@ zebra-chain = { path = "../zebra-chain" }
 zebra-network = { path = "../zebra-network" }
 eyre = "0.4.3"
 color-eyre = "0.3.2"
+zebra-state = { path = "../zebra-state" }
 
 [dev-dependencies]
 abscissa_core = { version = "0.5", features = ["testing"] }

zebrad/src/commands/connect.rs

@@ -42,25 +42,25 @@ impl Runnable for ConnectCmd {
 
 impl ConnectCmd {
     async fn connect(&self) -> Result<(), Report> {
-        use zebra_network::{Request, Response};
-
         info!("begin tower-based peer handling test stub");
         use tower::{buffer::Buffer, service_fn, Service, ServiceExt};
 
+        // The service that our node uses to respond to requests by peers
         let node = Buffer::new(
             service_fn(|req| async move {
                 info!(?req);
-                Ok::<Response, Report>(Response::Nil)
+                Ok::<zebra_network::Response, Report>(zebra_network::Response::Nil)
             }),
             1,
         );
 
         let mut config = app_config().network.clone();
         // Use a different listen addr so that we don't conflict with another local node.
-        config.listen_addr = "127.0.0.1:38233".parse().unwrap();
+        config.listen_addr = "127.0.0.1:38233".parse()?;
         // Connect only to the specified peer.
         config.initial_mainnet_peers.insert(self.addr.to_string());
 
+        let mut state = zebra_state::in_memory::init();
         let (mut peer_set, _address_book) = zebra_network::init(config, node).await;
         let mut retry_peer_set =
             tower::retry::Retry::new(zebra_network::RetryErrors, peer_set.clone());
@@ -81,19 +81,25 @@ impl ConnectCmd {
             177, 29, 170, 27, 145, 113, 132, 236, 232, 15, 4, 0,
         ]);
 
+        // TODO(jlusby): Replace with real state service
         let mut downloaded_block_heights = BTreeSet::<BlockHeight>::new();
         downloaded_block_heights.insert(BlockHeight(0));
+
         let mut block_requests = FuturesUnordered::new();
         let mut requested_block_heights = 0;
+
         while requested_block_heights < 700_000 {
             // Request the next 500 hashes.
-            retry_peer_set.ready_and().await.unwrap();
+            let hashes = if let Ok(zebra_network::Response::BlockHeaderHashes(hashes)) =
-            let hashes = if let Ok(Response::BlockHeaderHashes(hashes)) = retry_peer_set
+                retry_peer_set
-                .call(Request::FindBlocks {
+                    .ready_and()
-                    known_blocks: vec![tip],
+                    .await
-                    stop: None,
+                    .map_err(|e| eyre!(e))?
-                })
+                    .call(zebra_network::Request::FindBlocks {
-                .await
+                        known_blocks: vec![tip],
+                        stop: None,
+                    })
+                    .await
             {
                 info!(
                     new_hashes = hashes.len(),
@@ -113,17 +119,27 @@ impl ConnectCmd {
 
             // Request the corresponding blocks in chunks
             for chunk in hashes.chunks(10usize) {
-                peer_set.ready_and().await.unwrap();
+                let request = peer_set.ready_and().await.map_err(|e| eyre!(e))?.call(
-                block_requests
+                    zebra_network::Request::BlocksByHash(chunk.iter().cloned().collect()),
-                    .push(peer_set.call(Request::BlocksByHash(chunk.iter().cloned().collect())));
+                );
+
+                block_requests.push(request);
             }
 
             // Allow at most 300 block requests in flight.
             while block_requests.len() > 300 {
                 match block_requests.next().await {
-                    Some(Ok(Response::Blocks(blocks))) => {
+                    Some(Ok(zebra_network::Response::Blocks(blocks))) => {
-                        for block in &blocks {
+                        for block in blocks {
                             downloaded_block_heights.insert(block.coinbase_height().unwrap());
+                            let block = block.into();
+                            state
+                                .ready_and()
+                                .await
+                                .map_err(|e| eyre!(e))?
+                                .call(zebra_state::Request::AddBlock { block })
+                                .await
+                                .map_err(|e| eyre!(e))?;
                         }
                     }
                     Some(Err(e)) => {
@@ -134,9 +150,17 @@ impl ConnectCmd {
             }
         }
 
-        while let Some(Ok(Response::Blocks(blocks))) = block_requests.next().await {
+        while let Some(Ok(zebra_network::Response::Blocks(blocks))) = block_requests.next().await {
-            for block in &blocks {
+            for block in blocks {
                 downloaded_block_heights.insert(block.coinbase_height().unwrap());
+                let block = block.into();
+                state
+                    .ready_and()
+                    .await
+                    .map_err(|e| eyre!(e))?
+                    .call(zebra_state::Request::AddBlock { block })
+                    .await
+                    .map_err(|e| eyre!(e))?;
             }
         }