blockchain: Implement add() for BlockProposal.

src/blockchain2/blockstore.rs

@@ -12,8 +12,6 @@ pub struct BlockStore(sled::Tree);
 
 impl BlockStore {
     /// Opens a new or existing `BlockStore` on the given sled database.
-    /// The database is typically initialized as a global instance reference
-    /// with e.g. lazy_static.
     pub fn new(db: &sled::Db, genesis_ts: Timestamp, genesis_data: blake3::Hash) -> Result<Self> {
         let tree = db.open_tree(SLED_BLOCK_TREE)?;
         let store = Self(tree);
@@ -29,17 +27,19 @@ impl BlockStore {
     /// Insert a slice of [`Block`] into the blockstore. With sled, the
     /// operation is done as a batch.
     /// The blocks are hashed with BLAKE3 and this blockhash is used as
-    /// the key, where value is the serialized block itself.
-    pub fn insert(&self, blocks: &[Block]) -> Result<()> {
+    /// the key, while value is the serialized block itself.
+    pub fn insert(&self, blocks: &[Block]) -> Result<Vec<blake3::Hash>> {
+        let mut ret = Vec::with_capacity(blocks.len());
         let mut batch = Batch::default();
         for i in blocks {
             let serialized = serialize(i);
             let blockhash = blake3::hash(&serialized);
             batch.insert(blockhash.as_bytes(), serialized);
+            ret.push(blockhash);
         }
 
         self.0.apply_batch(batch)?;
-        Ok(())
+        Ok(ret)
     }
 
     /// Fetch given blockhashes from the blockstore.

src/blockchain2/metadatastore.rs

@@ -1,3 +1,5 @@
+use sled::Batch;
+
 use crate::{consensus2::StreamletMetadata, util::serial::serialize, Result};
 
 const SLED_STREAMLET_METADATA_TREE: &[u8] = b"_streamlet_metadata";
@@ -13,8 +15,15 @@ impl StreamletMetadataStore {
     /// Insert [`StreamletMetadata`] into the `MetadataStore`.
     /// The blockhash for the metadata is used as the key,
     /// where value is the serialized metadata.
-    pub fn insert(&self, block: blake3::Hash, metadata: &StreamletMetadata) -> Result<()> {
-        self.0.insert(block.as_bytes(), serialize(metadata))?;
+    pub fn insert(&self, blocks: &[blake3::Hash], metadatas: &[StreamletMetadata]) -> Result<()> {
+        assert_eq!(blocks.len(), metadatas.len());
+        let mut batch = Batch::default();
+
+        for (i, hash) in blocks.iter().enumerate() {
+            batch.insert(hash.as_bytes(), serialize(&metadatas[i]));
+        }
+
+        self.0.apply_batch(batch)?;
         Ok(())
     }
 }

src/blockchain2/mod.rs

@@ -1,7 +1,7 @@
 use std::io;
 
 use crate::{
-    consensus2::{block::BlockProposal, util::Timestamp},
+    consensus2::{block::BlockProposal, util::Timestamp, Block},
     impl_vec,
     util::serial::{Decodable, Encodable, ReadExt, VarInt, WriteExt},
     Result,
@@ -45,7 +45,24 @@ impl Blockchain {
 
     /// Batch insert [`BlockProposal`]s.
     pub fn add(&mut self, proposals: &[BlockProposal]) -> Result<Vec<blake3::Hash>> {
-        todo!()
+        // TODO: Engineer this function in a better way.
+        let mut ret = Vec::with_capacity(proposals.len());
+
+        for prop in proposals {
+            // Store transactions
+            let tx_hashes = self.transactions.insert(&prop.txs)?;
+
+            // Store block
+            let block =
+                Block { st: prop.st, sl: prop.sl, txs: tx_hashes, metadata: prop.metadata.clone() };
+            let blockhash = self.blocks.insert(&[block])?;
+            ret.push(blockhash[0]);
+
+            // Store streamlet metadata
+            self.streamlet_metadata.insert(&[blockhash[0]], &[prop.sm.clone()])?;
+        }
+
+        Ok(ret)
     }
 }
 

src/blockchain2/txstore.rs

@@ -1,12 +1,33 @@
-use crate::Result;
+use sled::Batch;
+
+use crate::{consensus2::Tx, util::serial::serialize, Result};
 
 const SLED_TX_TREE: &[u8] = b"_transactions";
 
 pub struct TxStore(sled::Tree);
 
 impl TxStore {
+    /// Opens a new or existing `TxStore` on the given sled database.
     pub fn new(db: &sled::Db) -> Result<Self> {
         let tree = db.open_tree(SLED_TX_TREE)?;
         Ok(Self(tree))
     }
+
+    /// Insert a slice of [`Tx`] into the txstore. With sled, the
+    /// operation is done as a batch.
+    /// The transactions are hashed with BLAKE3 and this hash is
+    /// used as the key, while value is the serialized tx itself.
+    pub fn insert(&self, txs: &[Tx]) -> Result<Vec<blake3::Hash>> {
+        let mut ret = Vec::with_capacity(txs.len());
+        let mut batch = Batch::default();
+        for i in txs {
+            let serialized = serialize(i);
+            let txhash = blake3::hash(&serialized);
+            batch.insert(txhash.as_bytes(), serialized);
+            ret.push(txhash);
+        }
+
+        self.0.apply_batch(batch)?;
+        Ok(ret)
+    }
 }