Added Fetch and Load Methods

blockchain/blocks/src/block.rs

@@ -22,9 +22,9 @@ struct PoStProof {}
 /// A complete block
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct Block {
-    header: BlockHeader,
-    bls_messages: Vec<UnsignedMessage>,
-    secp_messages: Vec<SignedMessage>,
+    pub header: BlockHeader,
+    pub bls_messages: Vec<UnsignedMessage>,
+    pub secp_messages: Vec<SignedMessage>,
 }
 
 impl Block {

blockchain/blocks/src/tipset.rs

@@ -3,7 +3,6 @@
 
 #![allow(unused_variables)]
 #![allow(dead_code)]
-
 use super::{Block, BlockHeader, Error, Ticket};
 use cid::Cid;
 use clock::ChainEpoch;
@@ -38,6 +37,10 @@ impl TipSetKeys {
         }
         true
     }
+    /// Returns tipset keys
+    pub fn tipset_keys(&self) -> &[Cid] {
+        &self.cids
+    }
 }
 
 impl ser::Serialize for TipSetKeys {

blockchain/chain/src/store/chain_store.rs

@@ -5,7 +5,8 @@ use super::{Error, TipIndex, TipSetMetadata};
 use blocks::{BlockHeader, Tipset};
 use cid::Cid;
 use db::{Error as DbError, Read, RocksDb as Blockstore, Write};
-use encoding::from_slice;
+use encoding::{de::DeserializeOwned, from_slice};
+use message::{SignedMessage, UnsignedMessage};
 use num_bigint::BigUint;
 use raw_block::RawBlock;
 use std::path::Path;
@@ -106,4 +107,51 @@ impl<'a> ChainStore<'a> {
     pub fn blockstore(&self) -> &Blockstore {
         &self.db
     }
+    /// Returns Tipset from key-value store from provided cids
+    pub fn tipset(&self, cids: &[Cid]) -> Result<Tipset, Error> {
+        let mut block_headers = Vec::new();
+        for c in cids {
+            let raw_header = self.db.read(c.key())?;
+            if let Some(x) = raw_header {
+                // decode raw header into BlockHeader
+                let bh = from_slice(&x)?;
+                block_headers.push(bh);
+            } else {
+                return Err(Error::KeyValueStore(
+                    "Key for header does not exist".to_string(),
+                ));
+            }
+        }
+        // construct new Tipset to return
+        let ts = Tipset::new(block_headers)?;
+        Ok(ts)
+    }
+
+    /// Returns a Tuple of bls messages of type UnsignedMessage and secp messages
+    /// of type SignedMessage
+    pub fn messages(
+        &self,
+        _bh: &BlockHeader,
+    ) -> Result<(Vec<UnsignedMessage>, Vec<SignedMessage>), Error> {
+        // TODO dependent on HAMT
+
+        let bls_msgs: Vec<UnsignedMessage> = self.messages_from_cids(Vec::new())?;
+        let secp_msgs: Vec<SignedMessage> = self.messages_from_cids(Vec::new())?;
+        Ok((bls_msgs, secp_msgs))
+    }
+    /// Returns messages from key-value store
+    pub fn messages_from_cids<T>(&self, keys: Vec<&Cid>) -> Result<Vec<T>, Error>
+    where
+        T: DeserializeOwned,
+    {
+        keys.iter()
+            .map(|k| {
+                let value = self.db.read(&k.key())?;
+                let bytes = value.ok_or_else(|| Error::UndefinedKey(k.to_string()))?;
+
+                // Decode bytes into type T
+                from_slice(&bytes)?
+            })
+            .collect()
+    }
 }

blockchain/sync_manager/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2018"
 
 [dependencies]
 address = { package = "forest_address", path = "../../vm/address" }
+amt = { package = "ipld_amt", path = "../../ipld/amt" }
 blocks = { package = "forest_blocks", path = "../blocks" }
 db = { path = "../../node/db" }
 encoding = { package = "forest_encoding", path = "../../encoding" }

blockchain/sync_manager/src/errors.rs

@@ -1,6 +1,7 @@
 // Copyright 2020 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+use amt::Error as AmtErr;
 use blocks::Error as BlkErr;
 use chain::Error as StoreErr;
 use cid::Error as CidErr;
@@ -23,6 +24,8 @@ pub enum Error {
     Store(String),
     /// Error originating from key-value store
     KeyValueStore(String),
+    /// Error originating from the AMT
+    AMT(String),
 }
 
 impl fmt::Display for Error {
@@ -41,6 +44,7 @@ impl fmt::Display for Error {
             Error::Encoding(msg) => write!(f, "Error originating from Encoding type: {}", msg),
             Error::InvalidCid(msg) => write!(f, "Error originating from CID construction: {}", msg),
             Error::Store(msg) => write!(f, "Error originating from ChainStore: {}", msg),
+            Error::AMT(msg) => write!(f, "Error originating from the AMT: {}", msg),
         }
     }
 }
@@ -80,3 +84,9 @@ impl From<StoreErr> for Error {
         Error::Store(e.to_string())
     }
 }
+
+impl From<AmtErr> for Error {
+    fn from(e: AmtErr) -> Error {
+        Error::AMT(e.to_string())
+    }
+}

blockchain/sync_manager/src/sync.rs

@@ -5,11 +5,12 @@
 
 use super::errors::Error;
 use super::manager::SyncManager;
-use blocks::{Block, FullTipset, Tipset};
+use amt::{BlockStore, AMT};
+use blocks::{Block, FullTipset, TipSetKeys, Tipset};
 use chain::ChainStore;
-use cid::{Cid, Codec, Error as CidError, Version};
+use cid::{Cid, Error as CidError};
 use libp2p::core::PeerId;
-use multihash::Multihash;
+use message::MsgMeta;
 use raw_block::RawBlock;
 
 pub struct Syncer<'a> {
@@ -58,7 +59,8 @@ impl<'a> Syncer<'a> {
         // incoming tipset from miners does not appear to be better than our best chain, ignoring for now
         Ok(())
     }
-
+    /// Validates message root from header matches message root generated from the
+    /// bls and secp messages contained in the passed in block and stores them in a key-value store
     fn validate_msg_data(&self, block: &Block) -> Result<(), Error> {
         let sm_root = self.compute_msg_data(block)?;
         // TODO change message_receipts to messages() once #192 is in
@@ -71,17 +73,54 @@ impl<'a> Syncer<'a> {
 
         Ok(())
     }
-    fn compute_msg_data(&self, block: &Block) -> Result<Cid, CidError> {
-        // TODO compute message roots
+    /// Returns message root CID from bls and secp message contained in the param Block
+    fn compute_msg_data(&self, block: &Block) -> Result<Cid, Error> {
+        // collect bls and secp cids
+        let bls_cids = cids_from_messages(block.bls_msgs())?;
+        let secp_cids = cids_from_messages(block.secp_msgs())?;
+        // generate AMT and batch set message values
+        let bls_root = AMT::new_from_slice(self.chain_store.blockstore(), &bls_cids)?;
+        let secp_root = AMT::new_from_slice(self.chain_store.blockstore(), &secp_cids)?;
 
-        let _bls_cids = cids_from_messages(block.bls_msgs())?;
-        let _secp_cids = cids_from_messages(block.secp_msgs())?;
+        let meta = MsgMeta {
+            bls_message_root: bls_root,
+            secp_message_root: secp_root,
+        };
+        // store message roots and receive meta_root
+        let meta_root = self.chain_store.blockstore().put(&meta)?;
 
-        // TODO temporary until AMT structure is implemented
-        // see Lotus implementation https://github.com/filecoin-project/lotus/blob/master/chain/sync.go#L338
-        // will return a new CID representing both message roots
-        let hash = Multihash::from_bytes(vec![0, 0]);
-        Ok(Cid::new(Codec::DagCBOR, Version::V1, hash.unwrap()))
+        Ok(meta_root)
+    }
+    /// Returns FullTipset from store if TipSetKeys exist in key-value store otherwise requests FullTipset
+    /// from block sync
+    fn fetch_tipsets(&self, _peer_id: PeerId, tsk: TipSetKeys) -> Result<FullTipset, Error> {
+        let fts = match self.load_fts(tsk) {
+            Ok(fts) => fts,
+            // TODO call into block sync to request FullTipset -> self.blocksync.get_full_tipset(_peer_id, tsk)
+            Err(e) => return Err(e), // blocksync
+        };
+        Ok(fts)
+    }
+    /// Returns a reconstructed FullTipset from store if keys exist
+    fn load_fts(&self, keys: TipSetKeys) -> Result<FullTipset, Error> {
+        let mut blocks = Vec::new();
+        // retrieve tipset from store based on passed in TipSetKeys
+        let ts = self.chain_store.tipset(keys.tipset_keys())?;
+        for header in ts.blocks() {
+            // retrieve bls and secp messages from specified BlockHeader
+            let (bls_msgs, secp_msgs) = self.chain_store.messages(&header)?;
+            // construct a full block
+            let full_block = Block {
+                header: header.clone(),
+                bls_messages: bls_msgs,
+                secp_messages: secp_msgs,
+            };
+            // push vector of full blocks to build FullTipset
+            blocks.push(full_block);
+        }
+        // construct FullTipset
+        let fts = FullTipset::new(blocks);
+        Ok(fts)
     }
 }
 

ipld/cid/Cargo.toml

@@ -9,4 +9,4 @@ multihash = "0.9.4"
 multibase = "0.7.0"
 integer-encoding = "1.0.3"
 encoding = { package = "forest_encoding", path = "../../encoding" }
-serde = { version = "1.0", features = ["derive"] }
+serde = { version = "1.0", features = ["derive"] }

ipld/cid/src/error.rs

@@ -5,7 +5,6 @@ use encoding::Error as EncodingError;
 use multibase;
 use multihash;
 use std::{error, fmt, io};
-
 /// Error types
 #[derive(PartialEq, Eq, Clone, Debug)]
 pub enum Error {

vm/message/src/lib.rs

@@ -10,6 +10,11 @@ pub use signed_message::*;
 pub use unsigned_message::*;
 
 use address::Address;
+use cid::Cid;
+use encoding::{
+    de::{self, Deserialize, Deserializer},
+    ser::{self, Serializer},
+};
 use num_bigint::BigUint;
 use vm::{MethodNum, Serialized, TokenAmount};
 
@@ -31,3 +36,34 @@ pub trait Message {
     /// Returns the gas limit for the message
     fn gas_limit(&self) -> &BigUint;
 }
+
+pub struct MsgMeta {
+    pub bls_message_root: Cid,
+    pub secp_message_root: Cid,
+}
+
+impl ser::Serialize for MsgMeta {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        (
+            self.bls_message_root.clone(),
+            self.secp_message_root.clone(),
+        )
+            .serialize(serializer)
+    }
+}
+
+impl<'de> de::Deserialize<'de> for MsgMeta {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let (bls_message_root, secp_message_root) = Deserialize::deserialize(deserializer)?;
+        Ok(Self {
+            bls_message_root,
+            secp_message_root,
+        })
+    }
+}