fix: post-toolchain and snarkVM update alignment

Signed-off-by: ljedrz <ljedrz@users.noreply.github.com>

Author: ljedrz

node/bft/src/gateway.rs

@@ -1072,10 +1072,7 @@ impl<N: Network> Reading for Gateway<N> {
     type Message = Event<N>;
 
     /// The maximum queue depth of incoming messages for a single peer.
-    const MESSAGE_QUEUE_DEPTH: usize = 2
-        * BatchHeader::<N>::MAX_GC_ROUNDS
-        * N::LATEST_MAX_CERTIFICATES().unwrap() as usize
-        * BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH;
+    const MESSAGE_QUEUE_DEPTH: usize = 256_000;
 
     /// Creates a [`Decoder`] used to interpret messages from the network.
     /// The `side` param indicates the connection side **from the node's perspective**.
@@ -1107,10 +1104,7 @@ impl<N: Network> Writing for Gateway<N> {
     type Message = Event<N>;
 
     /// The maximum queue depth of outgoing messages for a single peer.
-    const MESSAGE_QUEUE_DEPTH: usize = 2
-        * BatchHeader::<N>::MAX_GC_ROUNDS
-        * N::LATEST_MAX_CERTIFICATES().unwrap() as usize
-        * BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH;
+    const MESSAGE_QUEUE_DEPTH: usize = 256_000;
 
     /// Creates an [`Encoder`] used to write the outbound messages to the target stream.
     /// The `side` parameter indicates the connection side **from the node's perspective**.
@@ -1254,7 +1248,7 @@ impl<N: Network> Gateway<N> {
         peer_ip: Option<SocketAddr>,
         restrictions_id: Field<N>,
         stream: &'a mut TcpStream,
-    ) -> io::Result<(SocketAddr, Framed<&mut TcpStream, EventCodec<N>>)> {
+    ) -> io::Result<(SocketAddr, Framed<&'a mut TcpStream, EventCodec<N>>)> {
         // This value is immediately guaranteed to be present, so it can be unwrapped.
         let peer_ip = peer_ip.unwrap();
 
@@ -1319,7 +1313,7 @@ impl<N: Network> Gateway<N> {
         peer_ip: &mut Option<SocketAddr>,
         restrictions_id: Field<N>,
         stream: &'a mut TcpStream,
-    ) -> io::Result<(SocketAddr, Framed<&mut TcpStream, EventCodec<N>>)> {
+    ) -> io::Result<(SocketAddr, Framed<&'a mut TcpStream, EventCodec<N>>)> {
         // Construct the stream.
         let mut framed = Framed::new(stream, EventCodec::<N>::handshake());
 
@@ -1453,7 +1447,10 @@ mod prop_tests {
     use snarkos_account::Account;
     use snarkos_node_bft_ledger_service::MockLedgerService;
     use snarkos_node_bft_storage_service::BFTMemoryService;
-    use snarkos_node_tcp::P2P;
+    use snarkos_node_tcp::{
+        P2P,
+        protocols::{Reading, Writing},
+    };
     use snarkvm::{
         ledger::{
             committee::{
@@ -1692,4 +1689,15 @@ mod prop_tests {
             }
         }
     }
+
+    #[test]
+    fn ensure_sufficient_rw_queue_depth() {
+        let desired_rw_queue_depth = 2
+            * BatchHeader::<MainnetV0>::MAX_GC_ROUNDS
+            * MainnetV0::LATEST_MAX_CERTIFICATES().unwrap() as usize
+            * BatchHeader::<MainnetV0>::MAX_TRANSMISSIONS_PER_BATCH;
+
+        assert!(<Gateway<MainnetV0> as Reading>::MESSAGE_QUEUE_DEPTH >= desired_rw_queue_depth);
+        assert!(<Gateway<MainnetV0> as Writing>::MESSAGE_QUEUE_DEPTH >= desired_rw_queue_depth);
+    }
 }

node/bft/src/helpers/pending.rs

@@ -15,7 +15,7 @@
 
 use crate::MAX_FETCH_TIMEOUT_IN_MS;
 use snarkos_node_bft_ledger_service::LedgerService;
-use snarkvm::console::network::{consensus_config_value, Network};
+use snarkvm::console::network::{Network, consensus_config_value};
 
 use parking_lot::RwLock;
 use std::{
@@ -34,14 +34,14 @@ pub(crate) const CALLBACK_EXPIRATION_IN_SECS: i64 = MAX_FETCH_TIMEOUT_IN_MS.div_
 /// Returns the maximum number of redundant requests for the number of validators in the specified round.
 pub fn max_redundant_requests<N: Network>(ledger: Arc<dyn LedgerService<N>>, round: u64) -> usize {
     // Determine the number of validators in the committee lookback for the given round.
-    let num_validators = ledger
-        .get_committee_lookback_for_round(round)
-        .map(|committee| committee.num_members())
-        .ok()
-        .unwrap_or_else(|| {
-            let max_committee_size = consensus_config_value!(N, MAX_CERTIFICATES, ledger.latest_block_height()).unwrap();
-            max_committee_size as usize
-     });
+    let num_validators =
+        ledger.get_committee_lookback_for_round(round).map(|committee| committee.num_members()).ok().unwrap_or_else(
+            || {
+                let max_committee_size =
+                    consensus_config_value!(N, MAX_CERTIFICATES, ledger.latest_block_height()).unwrap();
+                max_committee_size as usize
+            },
+        );
 
     // Note: It is adequate to set this value to the availability threshold,
     // as with high probability one will respond honestly (in the best and worst case

node/bft/storage-service/src/persistent.rs

@@ -61,10 +61,9 @@ impl<N: Network> BFTPersistentStorage<N> {
     /// Initializes a new BFT persistent storage service.
     pub fn open(storage_mode: StorageMode) -> Result<Self> {
         let max_committee_size = N::LATEST_MAX_CERTIFICATES().unwrap();
-        let capacity = NonZeroUsize::new(
-            (max_committee_size as usize) * (BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH) * 2,
-        )
-        .ok_or_else(|| anyhow!("Could not construct NonZeroUsize"))?;
+        let capacity =
+            NonZeroUsize::new((max_committee_size as usize) * (BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH) * 2)
+                .ok_or_else(|| anyhow!("Could not construct NonZeroUsize"))?;
 
         Ok(Self {
             transmissions: internal::RocksDB::open_map(N::ID, storage_mode.clone(), MapID::BFT(BFTMap::Transmissions))?,
@@ -82,10 +81,9 @@ impl<N: Network> BFTPersistentStorage<N> {
     #[cfg(any(test, feature = "test"))]
     pub fn open_testing(temp_dir: std::path::PathBuf, dev: Option<u16>) -> Result<Self> {
         let max_committee_size = N::MAX_CERTIFICATES.last().unwrap().1;
-        let capacity = NonZeroUsize::new(
-            (max_committee_size as usize) * (BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH) * 2,
-        )
-        .ok_or_else(|| anyhow!("Could not construct NonZeroUsize"))?;
+        let capacity =
+            NonZeroUsize::new((max_committee_size as usize) * (BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH) * 2)
+                .ok_or_else(|| anyhow!("Could not construct NonZeroUsize"))?;
 
         Ok(Self {
             transmissions: internal::RocksDB::open_map_testing(

node/bft/tests/components/pending.rs

@@ -15,16 +15,16 @@
 
 use crate::common::{CurrentNetwork, primary::new_test_committee, utils::sample_ledger};
 use snarkos_node_bft::helpers::max_redundant_requests;
-use snarkvm::{ledger::committee::Committee, prelude::TestRng};
+use snarkvm::prelude::{Network, TestRng};
 
 #[test]
 fn test_max_redundant_requests() {
-    const NUM_NODES: u16 = Committee::<CurrentNetwork>::MAX_COMMITTEE_SIZE;
+    let num_nodes: u16 = CurrentNetwork::LATEST_MAX_CERTIFICATES().unwrap();
 
     // Initialize the RNG.
     let mut rng = TestRng::default();
     // Initialize the accounts and the committee.
-    let (accounts, committee) = new_test_committee(NUM_NODES, &mut rng);
+    let (accounts, committee) = new_test_committee(num_nodes, &mut rng);
     // Sample a ledger.
     let ledger = sample_ledger(&accounts, &committee, &mut rng);
     // Ensure the maximum number of redundant requests is correct and consistent across iterations.

node/bft/tests/components/worker.rs

@@ -20,7 +20,7 @@ use crate::common::{
 };
 use snarkos_node_bft::helpers::max_redundant_requests;
 use snarkvm::{
-    ledger::{committee::Committee, narwhal::TransmissionID},
+    ledger::narwhal::TransmissionID,
     prelude::{Network, TestRng},
 };
 
@@ -29,12 +29,12 @@ use std::net::SocketAddr;
 #[tokio::test]
 #[rustfmt::skip]
 async fn test_resend_transmission_request() {
-    const NUM_NODES: u16 = Committee::<CurrentNetwork>::MAX_COMMITTEE_SIZE;
+    let num_nodes: u16 = CurrentNetwork::LATEST_MAX_CERTIFICATES().unwrap();
 
     // Initialize the RNG.
     let mut rng = TestRng::default();
     // Initialize the accounts and the committee.
-    let (accounts, committee) = new_test_committee(NUM_NODES, &mut rng);
+    let (accounts, committee) = new_test_committee(num_nodes, &mut rng);
     // Sample a ledger.
     let ledger = sample_ledger(&accounts, &committee, &mut rng);
     // Sample a worker.
@@ -111,12 +111,12 @@ async fn test_resend_transmission_request() {
 #[tokio::test]
 #[rustfmt::skip]
 async fn test_flood_transmission_requests() {
-    const NUM_NODES: u16 = Committee::<CurrentNetwork>::MAX_COMMITTEE_SIZE;
+    let num_nodes: u16 = CurrentNetwork::LATEST_MAX_CERTIFICATES().unwrap();
 
     // Initialize the RNG.
     let mut rng = TestRng::default();
     // Initialize the accounts and the committee.
-    let (accounts, committee) = new_test_committee(NUM_NODES, &mut rng);
+    let (accounts, committee) = new_test_committee(num_nodes, &mut rng);
     // Sample a ledger.
     let ledger = sample_ledger(&accounts, &committee, &mut rng);
     // Sample a worker.

node/router/src/handshake.rs

@@ -90,7 +90,7 @@ impl<N: Network> Router<N> {
         peer_side: ConnectionSide,
         genesis_header: Header<N>,
         restrictions_id: Field<N>,
-    ) -> io::Result<(SocketAddr, Framed<&mut TcpStream, MessageCodec<N>>)> {
+    ) -> io::Result<(SocketAddr, Framed<&'a mut TcpStream, MessageCodec<N>>)> {
         // If this is an inbound connection, we log it, but don't know the listening address yet.
         // Otherwise, we can immediately register the listening address.
         let mut peer_ip = if peer_side == ConnectionSide::Initiator {
@@ -146,7 +146,7 @@ impl<N: Network> Router<N> {
         stream: &'a mut TcpStream,
         genesis_header: Header<N>,
         restrictions_id: Field<N>,
-    ) -> io::Result<(SocketAddr, Framed<&mut TcpStream, MessageCodec<N>>)> {
+    ) -> io::Result<(SocketAddr, Framed<&'a mut TcpStream, MessageCodec<N>>)> {
         // This value is immediately guaranteed to be present, so it can be unwrapped.
         let peer_ip = peer_ip.unwrap();
         // Construct the stream.
@@ -224,7 +224,7 @@ impl<N: Network> Router<N> {
         stream: &'a mut TcpStream,
         genesis_header: Header<N>,
         restrictions_id: Field<N>,
-    ) -> io::Result<(SocketAddr, Framed<&mut TcpStream, MessageCodec<N>>)> {
+    ) -> io::Result<(SocketAddr, Framed<&'a mut TcpStream, MessageCodec<N>>)> {
         // Construct the stream.
         let mut framed = Framed::new(stream, MessageCodec::<N>::handshake());