Improve documentation for consensus crate and remove (potentially dangerous) unwrap()

node/bft/src/bft.rs

@@ -80,7 +80,7 @@ pub struct BFT<N: Network> {
     leader_certificate_timer: Arc<AtomicI64>,
     /// The consensus sender.
     consensus_sender: Arc<OnceCell<ConsensusSender<N>>>,
-    /// The spawned handles.
+    /// Handles for all spawned tasks.
     handles: Arc<Mutex<Vec<JoinHandle<()>>>>,
     /// The BFT lock.
     lock: Arc<TMutex<()>>,
@@ -109,6 +109,9 @@ impl<N: Network> BFT<N> {
     }
 
     /// Run the BFT instance.
+    ///
+    /// This will return as soon as all required tasks are spawned.
+    /// The function must not be called more than once per instance.
     pub async fn run(
         &mut self,
         consensus_sender: Option<ConsensusSender<N>>,

node/bft/src/helpers/ready.rs

@@ -25,6 +25,7 @@ use snarkvm::{
 use indexmap::{IndexMap, IndexSet};
 use std::collections::{HashMap, VecDeque, hash_map::Entry::Vacant};
 
+/// Maintains a queue of verified ("ready") transmissions.
 #[derive(Clone, Debug)]
 pub struct Ready<N: Network> {
     /// Maps each transmission ID to its logical index (physical index + offset)

node/bft/src/primary.rs

@@ -1133,6 +1133,13 @@ impl<N: Network> Primary<N> {
 
 impl<N: Network> Primary<N> {
     /// Starts the primary handlers.
+    ///
+    /// For each receiver in the `primary_receiver` struct, there will be a dedicated tasks
+    /// that awaits new data and handles it accordingly.
+    /// Additionally, this spawns a task that periodically issues PrimaryPings and one that periodically
+    /// tries to move the the next round of batches.
+    ///
+    /// This function is called exactly once, in `Self::run()`.
     fn start_handlers(&self, primary_receiver: PrimaryReceiver<N>) {
         let PrimaryReceiver {
             mut rx_batch_propose,
@@ -1143,7 +1150,7 @@ impl<N: Network> Primary<N> {
             mut rx_unconfirmed_transaction,
         } = primary_receiver;
 
-        // Start the primary ping.
+        // Start the primary ping sender.
         let self_ = self.clone();
         self.spawn(async move {
             loop {
@@ -1277,7 +1284,7 @@ impl<N: Network> Primary<N> {
             }
         });
 
-        // Process the proposed batch.
+        // Start the proposed batch handler.
         let self_ = self.clone();
         self.spawn(async move {
             while let Some((peer_ip, batch_propose)) = rx_batch_propose.recv().await {
@@ -1298,7 +1305,7 @@ impl<N: Network> Primary<N> {
             }
         });
 
-        // Process the batch signature.
+        // Start the batch signature handler.
         let self_ = self.clone();
         self.spawn(async move {
             while let Some((peer_ip, batch_signature)) = rx_batch_signature.recv().await {
@@ -1319,7 +1326,7 @@ impl<N: Network> Primary<N> {
             }
         });
 
-        // Process the certified batch.
+        // Start the certified batch handler.
         let self_ = self.clone();
         self.spawn(async move {
             while let Some((peer_ip, batch_certificate)) = rx_batch_certified.recv().await {
@@ -1346,7 +1353,8 @@ impl<N: Network> Primary<N> {
             }
         });
 
-        // Periodically try to increment to the next round.
+        // This task periodically tries to move to the next round.
+        //
         // Note: This is necessary to ensure that the primary is not stuck on a previous round
         // despite having received enough certificates to advance to the next round.
         let self_ = self.clone();
@@ -1385,7 +1393,7 @@ impl<N: Network> Primary<N> {
             }
         });
 
-        // Process the unconfirmed solutions.
+        // Start a handler to process new unconfirmed solutions.
         let self_ = self.clone();
         self.spawn(async move {
             while let Some((solution_id, solution, callback)) = rx_unconfirmed_solution.recv().await {
@@ -1411,7 +1419,7 @@ impl<N: Network> Primary<N> {
             }
         });
 
-        // Process the unconfirmed transactions.
+        // Start a handler to process new unconfirmed transactions.
         let self_ = self.clone();
         self.spawn(async move {
             while let Some((transaction_id, transaction, callback)) = rx_unconfirmed_transaction.recv().await {

node/bft/src/worker.rs

@@ -42,6 +42,8 @@ use rand::seq::IteratorRandom;
 use std::{future::Future, net::SocketAddr, sync::Arc, time::Duration};
 use tokio::{sync::oneshot, task::JoinHandle, time::timeout};
 
+/// A worker's main role is maintaining a queue of verified ("ready") transmissions,
+/// which will eventually be fetched by the primary when the primary generates a new batch.
 #[derive(Clone)]
 pub struct Worker<N: Network> {
     /// The worker ID.
@@ -111,8 +113,6 @@ impl<N: Network> Worker<N> {
     /// The maximum number of transmissions allowed in a worker ping.
     pub const MAX_TRANSMISSIONS_PER_WORKER_PING: usize = BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH / 10;
 
-    // transmissions
-
     /// Returns the number of transmissions in the ready queue.
     pub fn num_transmissions(&self) -> usize {
         self.ready.read().num_transmissions()

node/consensus/README.md

@@ -6,5 +6,5 @@
 
 The `snarkos-node-consensus` crate provides the consensus layer for the snarkOS node.
 
-It builds on top of the `snarkos-node-bft` crate, which provides an abstract implementationv of the AleoBFT.
-More concretely, this crate provides the communication logic that allows multiple AleoBFT nodes to interact with each other.
+The crate builds on top of the `snarkos-node-bft`, which implements AleoBFT.
+It manages a ratelimiter/mempool for incoming transmissions, and manages construction of blocks from batches that have been confirmed by the BFT layer.