Add Drand Beacon Cache

blockchain/beacon/src/drand.rs

@@ -15,6 +15,7 @@ use grpc::ClientStub;
 use grpc::RequestOptions;
 use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize};
 use sha2::Digest;
+use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::error;
 use std::sync::Arc;
@@ -40,7 +41,7 @@ where
 {
     /// Verify a new beacon entry against the most recent one before it.
     fn verify_entry(
-        &self,
+        &mut self,
         curr: &BeaconEntry,
         prev: &BeaconEntry,
     ) -> Result<bool, Box<dyn error::Error>>;
@@ -59,6 +60,9 @@ pub struct DrandBeacon {
     drand_gen_time: u64,
     fil_gen_time: u64,
     fil_round_time: u64,
+
+    // Keeps track of computed beacon entries.
+    local_cache: HashMap<u64, BeaconEntry>,
 }
 
 impl DrandBeacon {
@@ -95,6 +99,7 @@ impl DrandBeacon {
             drand_gen_time: group.genesis_time,
             fil_round_time: interval,
             fil_gen_time: genesis_ts,
+            local_cache: HashMap::new(),
         })
     }
 }
@@ -103,7 +108,7 @@ impl DrandBeacon {
 #[async_trait]
 impl Beacon for DrandBeacon {
     fn verify_entry(
-        &self,
+        &mut self,
         curr: &BeaconEntry,
         prev: &BeaconEntry,
     ) -> Result<bool, Box<dyn error::Error>> {
@@ -123,21 +128,29 @@ impl Beacon for DrandBeacon {
         // Signature
         let sig = Signature::from_bytes(curr.data())?;
         let sig_match = bls_signatures::verify(&sig, &[digest], &[self.pub_key.key()]);
-        // TODO: Cache this result
+
+        // Cache the result
+        if sig_match && !self.local_cache.contains_key(&curr.round()) {
+            self.local_cache.insert(curr.round(), curr.clone());
+        }
         Ok(sig_match)
     }
 
     async fn entry(&self, round: u64) -> Result<BeaconEntry, Box<dyn error::Error>> {
-        // TODO: Cache values into a database
-        let mut req = PublicRandRequest::new();
-        req.round = round;
-        let resp = self
-            .client
-            .public_rand(grpc::RequestOptions::new(), req)
-            .drop_metadata()
-            .await?;
+        match self.local_cache.get(&round) {
+            Some(cached_entry) => Ok(cached_entry.clone()),
+            None => {
+                let mut req = PublicRandRequest::new();
+                req.round = round;
+                let resp = self
+                    .client
+                    .public_rand(grpc::RequestOptions::new(), req)
+                    .drop_metadata()
+                    .await?;
 
-        Ok(BeaconEntry::new(resp.round, resp.signature))
+                Ok(BeaconEntry::new(resp.round, resp.signature))
+            }
+        }
     }
 
     fn max_beacon_round_for_epoch(&self, fil_epoch: ChainEpoch) -> u64 {

blockchain/beacon/src/mock_beacon.rs

@@ -29,7 +29,11 @@ impl MockBeacon {
 
 #[async_trait]
 impl Beacon for MockBeacon {
-    fn verify_entry(&self, curr: &BeaconEntry, prev: &BeaconEntry) -> Result<bool, Box<dyn Error>> {
+    fn verify_entry(
+        &mut self,
+        curr: &BeaconEntry,
+        prev: &BeaconEntry,
+    ) -> Result<bool, Box<dyn Error>> {
         let oe = Self::entry_for_index(prev.round());
         Ok(oe.data() == curr.data())
     }

blockchain/beacon/tests/drand.rs

@@ -26,7 +26,7 @@ async fn construct_drand_beacon() {
 #[ignore]
 #[async_std::test]
 async fn ask_and_verify_beacon_entry() {
-    let beacon = new_beacon().await;
+    let mut beacon = new_beacon().await;
 
     let e2 = beacon.entry(2).await.unwrap();
     let e3 = beacon.entry(3).await.unwrap();
@@ -36,7 +36,7 @@ async fn ask_and_verify_beacon_entry() {
 #[ignore]
 #[async_std::test]
 async fn ask_and_verify_beacon_entry_fail() {
-    let beacon = new_beacon().await;
+    let mut beacon = new_beacon().await;
 
     let e2 = beacon.entry(2).await.unwrap();
     let e3 = beacon.entry(3).await.unwrap();

blockchain/blocks/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 features = ["json"]
 
 [dependencies]
+async-std = { version = "1.6.0", features = ["unstable"] }
 address = { package = "forest_address", path = "../../vm/address" }
 beacon = { path = "../beacon" }
 byteorder = "1.3.4"

blockchain/blocks/src/header/mod.rs

@@ -3,6 +3,7 @@
 
 use super::{Error, Ticket, Tipset, TipsetKeys};
 use address::Address;
+use async_std::sync::RwLock;
 use beacon::{self, Beacon, BeaconEntry};
 use cid::{multihash::Blake2b256, Cid};
 use clock::ChainEpoch;
@@ -368,10 +369,10 @@ impl BlockHeader {
     /// Validates if the current header's Beacon entries are valid to ensure randomness was generated correctly
     pub async fn validate_block_drand<B: Beacon>(
         &self,
-        beacon: Arc<B>,
+        beacon: Arc<RwLock<B>>,
         prev_entry: BeaconEntry,
     ) -> Result<(), Error> {
-        let max_round = beacon.max_beacon_round_for_epoch(self.epoch);
+        let max_round = beacon.read().await.max_beacon_round_for_epoch(self.epoch);
         if max_round == prev_entry.round() {
             if !self.beacon_entries.is_empty() {
                 return Err(Error::Validation(format!(
@@ -390,21 +391,22 @@ impl BlockHeader {
                 last.round()
             )));
         }
-        self.beacon_entries.iter().try_fold(
-            &prev_entry,
-            |prev, curr| -> Result<&BeaconEntry, Error> {
-                if !beacon
-                    .verify_entry(curr, &prev)
-                    .map_err(|e| Error::Validation(e.to_string()))?
-                {
-                    return Err(Error::Validation(format!(
-                        "beacon entry was invalid: curr:{:?}, prev: {:?}",
-                        curr, prev
-                    )));
-                }
-                Ok(curr)
-            },
-        )?;
+
+        let mut prev = &prev_entry;
+        for curr in &self.beacon_entries {
+            if !beacon
+                .write()
+                .await
+                .verify_entry(&curr, &prev)
+                .map_err(|e| Error::Validation(e.to_string()))?
+            {
+                return Err(Error::Validation(format!(
+                    "beacon entry was invalid: curr:{:?}, prev: {:?}",
+                    curr, prev
+                )));
+            }
+            prev = &curr;
+        }
         Ok(())
     }
 }

blockchain/chain_sync/src/sync.rs

@@ -56,7 +56,7 @@ pub struct ChainSyncer<DB, TBeacon> {
     state: Arc<RwLock<SyncState>>,
 
     /// Drand randomness beacon
-    beacon: Arc<TBeacon>,
+    beacon: Arc<RwLock<TBeacon>>,
 
     /// manages retrieving and updates state objects
     state_manager: Arc<StateManager<DB>>,
@@ -100,7 +100,7 @@ where
 {
     pub fn new(
         chain_store: ChainStore<DB>,
-        beacon: Arc<TBeacon>,
+        beacon: Arc<RwLock<TBeacon>>,
         network_send: Sender<NetworkMessage>,
         network_rx: Receiver<NetworkEvent>,
         genesis: Tipset,
@@ -1017,7 +1017,7 @@ mod tests {
         let gen = dummy_header();
         chain_store.set_genesis(gen.clone()).unwrap();
 
-        let beacon = Arc::new(MockBeacon::new(Duration::from_secs(1)));
+        let beacon = Arc::new(RwLock::new(MockBeacon::new(Duration::from_secs(1))));
 
         let genesis_ts = Tipset::new(vec![gen]).unwrap();
         (

blockchain/chain_sync/src/sync/peer_test.rs

@@ -30,7 +30,7 @@ fn peer_manager_update() {
     chain_store.set_genesis(dummy_header.clone()).unwrap();
 
     let genesis_ts = Tipset::new(vec![dummy_header]).unwrap();
-    let beacon = Arc::new(MockBeacon::new(Duration::from_secs(1)));
+    let beacon = Arc::new(RwLock::new(MockBeacon::new(Duration::from_secs(1))));
     let cs = ChainSyncer::new(
         chain_store,
         beacon,

forest/src/daemon.rs

@@ -64,7 +64,7 @@ pub(super) async fn start(config: Config) {
     // Initialize ChainSyncer
     let chain_syncer = ChainSyncer::new(
         chain_store,
-        Arc::new(beacon),
+        Arc::new(RwLock::new(beacon)),
         network_send.clone(),
         network_rx,
         genesis,