Merge branch 'testnet2' into locking_tweaks

ledger/src/state/ledger.rs

@@ -143,33 +143,63 @@ impl<N: Network> LedgerState<N> {
             ledger.blocks.add_block(genesis)?;
         }
 
-        // Retrieve each block from genesis to validate state.
-        for block_height in 0..=latest_block_height {
-            // Validate the ledger root every 500 blocks.
-            if block_height % 500 == 0 || block_height == latest_block_height {
-                // Log the progress of the ledger root validation procedure.
-                let progress = (block_height as f64 / latest_block_height as f64 * 100f64) as u8;
-                debug!("Validating the ledger root up to block {} ({}%)", block_height, progress);
-
-                // Ensure the ledger roots match their expected block heights.
-                let expected_ledger_root = ledger.get_previous_ledger_root(block_height)?;
-                match ledger.ledger_roots.get(&expected_ledger_root)? {
-                    Some(height) => {
-                        if block_height != height {
-                            return Err(anyhow!("Ledger expected block {}, found block {}", block_height, height));
-                        }
-                    }
-                    None => return Err(anyhow!("Ledger is missing ledger root for block {}", block_height)),
+        // Iterate and append each block hash from genesis to tip to validate ledger state.
+        const INCREMENT: u32 = 500;
+        let mut start_block_height = 0u32;
+        while start_block_height < latest_block_height {
+            // Compute the end block height (inclusive) for this iteration.
+            let end_block_height = std::cmp::min(start_block_height.saturating_add(INCREMENT), latest_block_height);
+
+            // Perform a spot check that the block headers for this iteration exists in storage.
+            if start_block_height % 2 == 0 {
+                let block_headers = ledger.get_block_headers(start_block_height, end_block_height)?;
+                assert_eq!(end_block_height - start_block_height + 1, block_headers.len() as u32);
+            }
+
+            // Retrieve the block hashes.
+            let block_hashes = ledger.get_block_hashes(start_block_height, end_block_height)?;
+
+            // Split the block hashes into (last_block_hash, [start_block_hash, ..., penultimate_block_hash]).
+            if let Some((end_block_hash, block_hashes_excluding_last)) = block_hashes.split_last() {
+                // Add the block hashes (up to penultimate) to the ledger tree.
+                ledger.ledger_tree.write().add_all(block_hashes_excluding_last)?;
+
+                // Check 1 - Ensure the root of the ledger tree matches the one saved in the ledger roots map.
+                let ledger_root = ledger.get_previous_ledger_root(end_block_height)?;
+                if ledger_root != ledger.ledger_tree.read().root() {
+                    return Err(anyhow!("Ledger has incorrect ledger tree state at block {}", end_block_height));
                 }
 
-                // Ensure the ledger tree matches the state of ledger roots.
-                if expected_ledger_root != ledger.ledger_tree.read().root() {
-                    return Err(anyhow!("Ledger has incorrect ledger tree state at block {}", block_height));
+                // Check 2 - Ensure the saved block height corresponding to this ledger root matches the expected block height.
+                let candidate_height = match ledger.ledger_roots.get(&ledger_root)? {
+                    Some(candidate_height) => candidate_height,
+                    None => return Err(anyhow!("Ledger is missing ledger root for block {}", end_block_height)),
+                };
+                if end_block_height != candidate_height {
+                    return Err(anyhow!(
+                        "Ledger expected block {}, found block {}",
+                        end_block_height,
+                        candidate_height
+                    ));
                 }
+
+                // Add the last block hash to the ledger tree.
+                ledger.ledger_tree.write().add(end_block_hash)?;
             }
 
-            // Add the block hash to the ledger tree.
-            ledger.ledger_tree.write().add(&ledger.get_block_hash(block_height)?)?;
+            // Update the starting block height for the next iteration.
+            start_block_height = std::cmp::min(end_block_height.saturating_add(1), latest_block_height);
+
+            // Log the progress of the validation procedure.
+            let progress = (start_block_height as f64 / latest_block_height as f64 * 100f64) as u8;
+            debug!("Validating the ledger up to block {} ({}%)", start_block_height, progress);
+        }
+
+        // If this is new storage, the while loop above did not execute,
+        // and proceed to add the genesis block hash into the ledger tree.
+        if start_block_height == 0u32 {
+            // Add the genesis block hash to the ledger tree.
+            ledger.ledger_tree.write().add(&N::genesis_block().hash())?;
         }
 
         // Update the latest ledger state.

src/environment/mod.rs

@@ -67,7 +67,7 @@ pub trait Environment: 'static + Clone + Debug + Default + Send + Sync {
     /// before dropping the connection; it should be no greater than the `HEARTBEAT_IN_SECS`.
     const CONNECTION_TIMEOUT_IN_SECS: u64 = 1;
     /// The duration in seconds to sleep in between ping requests with a connected peer.
-    const PING_SLEEP_IN_SECS: u64 = 75;
+    const PING_SLEEP_IN_SECS: u64 = 90;
     /// The duration in seconds after which a connected peer is considered inactive or
     /// disconnected if no message has been received in the meantime.
     const RADIO_SILENCE_IN_SECS: u64 = 150; // 2.5 minutes
@@ -126,7 +126,7 @@ impl<N: Network> Environment for SyncNode<N> {
     type Network = N;
     const NODE_TYPE: NodeType = NodeType::Sync;
     const SYNC_NODES: [&'static str; 5] = ["127.0.0.1:4131", "127.0.0.1:4133", "127.0.0.1:4134", "127.0.0.1:4135", "127.0.0.1:4136"];
-    const MINIMUM_NUMBER_OF_PEERS: usize = 21;
+    const MINIMUM_NUMBER_OF_PEERS: usize = 35;
     const MAXIMUM_NUMBER_OF_PEERS: usize = 256;
 }
 
@@ -138,8 +138,8 @@ impl<N: Network> Environment for ClientTrial<N> {
     type Network = N;
     const NODE_TYPE: NodeType = NodeType::Client;
     const SYNC_NODES: [&'static str; 5] = ["144.126.219.193:4132", "165.232.145.194:4132", "143.198.164.241:4132", "188.166.7.13:4132", "167.99.40.226:4132"];
-    const MINIMUM_NUMBER_OF_PEERS: usize = 11;
-    const MAXIMUM_NUMBER_OF_PEERS: usize = 35;
+    const MINIMUM_NUMBER_OF_PEERS: usize = 21;
+    const MAXIMUM_NUMBER_OF_PEERS: usize = 51;
 }
 
 #[derive(Clone, Debug, Default)]
@@ -150,7 +150,7 @@ impl<N: Network> Environment for MinerTrial<N> {
     type Network = N;
     const NODE_TYPE: NodeType = NodeType::Miner;
     const SYNC_NODES: [&'static str; 5] = ["144.126.219.193:4132", "165.232.145.194:4132", "143.198.164.241:4132", "188.166.7.13:4132", "167.99.40.226:4132"];
-    const MINIMUM_NUMBER_OF_PEERS: usize = 11;
+    const MINIMUM_NUMBER_OF_PEERS: usize = 21;
     const MAXIMUM_NUMBER_OF_PEERS: usize = 35;
     const COINBASE_IS_PUBLIC: bool = true;
 }

src/network/ledger.rs

@@ -33,6 +33,7 @@ use anyhow::Result;
 use chrono::Utc;
 use std::{
     collections::{HashMap, HashSet},
+    hash::{Hash, Hasher},
     net::SocketAddr,
     path::Path,
     sync::{
@@ -75,6 +76,43 @@ pub enum LedgerRequest<N: Network> {
     UnconfirmedBlock(SocketAddr, Block<N>, ProverRouter<N>),
 }
 
+///
+/// A request for a block with the specified height and possibly a hash.
+///
+#[derive(Clone, Debug)]
+pub struct BlockRequest<N: Network> {
+    height: u32,
+    hash: Option<N::BlockHash>,
+}
+
+// The height is the primary key, so use only it for hashing purposes.
+impl<N: Network> PartialEq for BlockRequest<N> {
+    fn eq(&self, other: &Self) -> bool {
+        self.height == other.height
+    }
+}
+
+impl<N: Network> Eq for BlockRequest<N> {}
+
+// The k1 == k2 -> hash(k1) == hash(k2) rule must hold.
+impl<N: Network> Hash for BlockRequest<N> {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.height.hash(state);
+    }
+}
+
+impl<N: Network> From<u32> for BlockRequest<N> {
+    fn from(height: u32) -> Self {
+        Self { height, hash: None }
+    }
+}
+
+impl<N: Network> From<(u32, Option<N::BlockHash>)> for BlockRequest<N> {
+    fn from((height, hash): (u32, Option<N::BlockHash>)) -> Self {
+        Self { height, hash }
+    }
+}
+
 ///
 /// A ledger for a specific network on the node server.
 ///
@@ -93,7 +131,7 @@ pub struct Ledger<N: Network, E: Environment> {
     /// The map of each peer to their ledger state := (node_type, status, is_fork, latest_block_height, block_locators).
     peers_state: RwLock<HashMap<SocketAddr, Option<(NodeType, State, Option<bool>, u32, BlockLocators<N>)>>>,
     /// The map of each peer to their block requests := HashMap<(block_height, block_hash), timestamp>
-    block_requests: RwLock<HashMap<SocketAddr, HashMap<(u32, Option<N::BlockHash>), i64>>>,
+    block_requests: RwLock<HashMap<SocketAddr, HashMap<BlockRequest<N>, i64>>>,
     /// A lock to ensure methods that need to be mutually-exclusive are enforced.
     /// In this context, `update_ledger`, `add_block`, and `update_block_requests` must be mutually-exclusive.
     block_requests_lock: Mutex<()>,
@@ -181,7 +219,7 @@ impl<N: Network, E: Environment> Ledger<N, E> {
         match request {
             LedgerRequest::BlockResponse(peer_ip, block, prover_router) => {
                 // Remove the block request from the ledger.
-                if self.remove_block_request(peer_ip, block.height(), block.hash()).await {
+                if self.remove_block_request(peer_ip, block.height()).await {
                     // On success, process the block response.
                     self.add_block(block, &prover_router).await;
                     // Check if syncing with this peer is complete.
@@ -275,9 +313,9 @@ impl<N: Network, E: Environment> Ledger<N, E> {
             // Ensure the block height is not part of a block request in a fork.
             let mut is_forked_block = false;
             for requests in self.block_requests.read().await.values() {
-                for (block_height, block_hash) in requests.keys() {
+                for block_request in requests.keys() {
                     // If the block is part of a fork, then don't attempt to add it again.
-                    if block_height == &block.height() && block_hash.is_some() {
+                    if block_request.height == block.height() && block_request.hash.is_some() {
                         is_forked_block = true;
                         break;
                     }
@@ -846,9 +884,9 @@ impl<N: Network, E: Environment> Ledger<N, E> {
     ///
     async fn add_block_request(&self, peer_ip: SocketAddr, block_height: u32, block_hash: Option<N::BlockHash>) {
         // Ensure the block request does not already exist.
-        if !self.contains_block_request(peer_ip, block_height, block_hash).await {
+        if !self.contains_block_request(peer_ip, block_height).await {
             match self.block_requests.write().await.get_mut(&peer_ip) {
-                Some(requests) => match requests.insert((block_height, block_hash), Utc::now().timestamp()) {
+                Some(requests) => match requests.insert((block_height, block_hash).into(), Utc::now().timestamp()) {
                     None => debug!("Requesting block {} from {}", block_height, peer_ip),
                     Some(_old_request) => self.add_failure(peer_ip, format!("Duplicate block request for {}", peer_ip)).await,
                 },
@@ -860,9 +898,9 @@ impl<N: Network, E: Environment> Ledger<N, E> {
     ///
     /// Returns `true` if the block request for the given block height to the specified peer exists.
     ///
-    async fn contains_block_request(&self, peer_ip: SocketAddr, block_height: u32, block_hash: Option<N::BlockHash>) -> bool {
+    async fn contains_block_request(&self, peer_ip: SocketAddr, block_height: u32) -> bool {
         match self.block_requests.read().await.get(&peer_ip) {
-            Some(requests) => requests.contains_key(&(block_height, block_hash)) || requests.contains_key(&(block_height, None)),
+            Some(requests) => requests.contains_key(&block_height.into()),
             None => false,
         }
     }
@@ -871,15 +909,14 @@ impl<N: Network, E: Environment> Ledger<N, E> {
     /// Removes a block request for the given block height to the specified peer.
     /// On success, returns `true`, otherwise returns `false`.
     ///
-    async fn remove_block_request(&self, peer_ip: SocketAddr, block_height: u32, block_hash: N::BlockHash) -> bool {
+    async fn remove_block_request(&self, peer_ip: SocketAddr, block_height: u32) -> bool {
         // Ensure the block height corresponds to a requested block.
-        if !self.contains_block_request(peer_ip, block_height, Some(block_hash)).await {
+        if !self.contains_block_request(peer_ip, block_height).await {
             self.add_failure(peer_ip, "Received an invalid block response".to_string()).await;
             false
         } else {
             if let Some(requests) = self.block_requests.write().await.get_mut(&peer_ip) {
-                let is_success =
-                    requests.remove(&(block_height, Some(block_hash))).is_some() || requests.remove(&(block_height, None)).is_some();
+                let is_success = requests.remove(&block_height.into()).is_some();
                 match is_success {
                     true => return true,
                     false => {