Deduplicate blocksync

node/bft/examples/simple_node.rs

@@ -19,12 +19,15 @@ extern crate tracing;
 use snarkos_account::Account;
 use snarkos_node_bft::{
     BFT,
+    Gateway,
     MEMORY_POOL_PORT,
     Primary,
     helpers::{ConsensusReceiver, PrimarySender, Storage, init_consensus_channels, init_primary_channels},
 };
 use snarkos_node_bft_ledger_service::TranslucentLedgerService;
 use snarkos_node_bft_storage_service::BFTMemoryService;
+use snarkos_node_sync::BlockSync;
+use snarkos_node_tcp::P2P;
 use snarkvm::{
     console::{account::PrivateKey, algorithms::BHP256, types::Address},
     ledger::{
@@ -139,8 +142,10 @@ pub async fn start_bft(
     let (consensus_sender, consensus_receiver) = init_consensus_channels::<CurrentNetwork>();
     // Initialize the consensus receiver handler.
     consensus_handler(consensus_receiver);
+    let gateway = Gateway::new(account.clone(), storage.clone(), ledger.clone(), None, &[], None)?;
+    let block_sync = Arc::new(BlockSync::new(ledger.clone(), gateway.tcp().clone()));
     // Initialize the BFT instance.
-    let mut bft = BFT::<CurrentNetwork>::new(account, storage, ledger, ip, &trusted_validators, dev)?;
+    let mut bft = BFT::<CurrentNetwork>::new(block_sync, account, storage, ledger, ip, &trusted_validators, dev)?;
     // Run the BFT instance.
     bft.run(Some(consensus_sender), sender.clone(), receiver).await?;
     // Retrieve the BFT's primary.
@@ -177,7 +182,10 @@ pub async fn start_primary(
     // Initialize the trusted validators.
     let trusted_validators = trusted_validators(node_id, num_nodes, peers);
     // Initialize the primary instance.
-    let mut primary = Primary::<CurrentNetwork>::new(account, storage, ledger, ip, &trusted_validators, dev)?;
+    let gateway = Gateway::new(account.clone(), storage.clone(), ledger.clone(), None, &[], None).unwrap();
+    let block_sync = Arc::new(BlockSync::new(ledger.clone(), gateway.tcp().clone()));
+    let mut primary =
+        Primary::<CurrentNetwork>::new(block_sync, account, storage, ledger, ip, &trusted_validators, dev)?;
     // Run the primary instance.
     primary.run(None, sender.clone(), receiver).await?;
     // Handle OS signals.

node/bft/src/bft.rs

@@ -30,6 +30,7 @@ use crate::{
 };
 use snarkos_account::Account;
 use snarkos_node_bft_ledger_service::LedgerService;
+use snarkos_node_sync::BlockSync;
 use snarkvm::{
     console::account::Address,
     ledger::{
@@ -60,9 +61,9 @@ use tokio::{
 
 #[derive(Clone)]
 pub struct BFT<N: Network> {
-    /// The primary.
+    /// The primary for this node.
     primary: Primary<N>,
-    /// The DAG.
+    /// The DAG of batches from which we build the blockchain.
     dag: Arc<RwLock<DAG<N>>>,
     /// The batch certificate of the leader from the current even round, if one was present.
     leader_certificate: Arc<RwLock<Option<BatchCertificate<N>>>>,
@@ -79,6 +80,7 @@ pub struct BFT<N: Network> {
 impl<N: Network> BFT<N> {
     /// Initializes a new instance of the BFT.
     pub fn new(
+        block_sync: Arc<BlockSync<N>>,
         account: Account<N>,
         storage: Storage<N>,
         ledger: Arc<dyn LedgerService<N>>,
@@ -87,7 +89,7 @@ impl<N: Network> BFT<N> {
         dev: Option<u16>,
     ) -> Result<Self> {
         Ok(Self {
-            primary: Primary::new(account, storage, ledger, ip, trusted_validators, dev)?,
+            primary: Primary::new(block_sync, account, storage, ledger, ip, trusted_validators, dev)?,
             dag: Default::default(),
             leader_certificate: Default::default(),
             leader_certificate_timer: Default::default(),
@@ -892,10 +894,12 @@ impl<N: Network> BFT<N> {
 
 #[cfg(test)]
 mod tests {
-    use crate::{BFT, MAX_LEADER_CERTIFICATE_DELAY_IN_SECS, helpers::Storage};
+    use crate::{BFT, Gateway, MAX_LEADER_CERTIFICATE_DELAY_IN_SECS, helpers::Storage};
     use snarkos_account::Account;
     use snarkos_node_bft_ledger_service::MockLedgerService;
     use snarkos_node_bft_storage_service::BFTMemoryService;
+    use snarkos_node_sync::BlockSync;
+    use snarkos_node_tcp::P2P;
     use snarkvm::{
         console::account::{Address, PrivateKey},
         ledger::{
@@ -934,6 +938,20 @@ mod tests {
         (committee, account, ledger, storage)
     }
 
+    // Helper function to set up BFT for testing.
+    fn initialize_bft(
+        account: Account<CurrentNetwork>,
+        storage: Storage<CurrentNetwork>,
+        ledger: Arc<MockLedgerService<CurrentNetwork>>,
+    ) -> anyhow::Result<BFT<CurrentNetwork>> {
+        // Initialize the gateway.
+        let gateway = Gateway::new(account.clone(), storage.clone(), ledger.clone(), None, &[], None)?;
+        // Create the block synchronization logic.
+        let block_sync = Arc::new(BlockSync::new(ledger.clone(), gateway.tcp().clone()));
+        // Initialize the BFT.
+        BFT::new(block_sync, account.clone(), storage.clone(), ledger.clone(), None, &[], None)
+    }
+
     #[test]
     #[tracing_test::traced_test]
     fn test_is_leader_quorum_odd() -> Result<()> {
@@ -964,8 +982,8 @@ mod tests {
         let storage = Storage::new(ledger.clone(), Arc::new(BFTMemoryService::new()), 10);
         // Initialize the account.
         let account = Account::new(rng)?;
-        // Initialize the BFT.
-        let bft = BFT::new(account.clone(), storage.clone(), ledger.clone(), None, &[], None)?;
+        // Set up the BFT logic.
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
         assert!(bft.is_timer_expired());
         // Ensure this call succeeds on an odd round.
         let result = bft.is_leader_quorum_or_nonleaders_available(1);
@@ -999,8 +1017,8 @@ mod tests {
         assert_eq!(storage.current_round(), 1);
         assert_eq!(storage.max_gc_rounds(), 10);
 
-        // Initialize the BFT.
-        let bft = BFT::new(account, storage, ledger, None, &[], None)?;
+        // Set up the BFT logic.
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
         assert!(bft.is_timer_expired()); // 0 + 5 < now()
 
         // Store is at round 1, and we are checking for round 2.
@@ -1021,8 +1039,8 @@ mod tests {
         assert_eq!(storage.current_round(), 2);
         assert_eq!(storage.max_gc_rounds(), 10);
 
-        // Initialize the BFT.
-        let bft = BFT::new(account, storage, ledger, None, &[], None)?;
+        // Set up the BFT logic.
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
         assert!(bft.is_timer_expired()); // 0 + 5 < now()
 
         // Ensure this call fails on an even round.
@@ -1061,8 +1079,11 @@ mod tests {
         let storage = Storage::new(ledger.clone(), Arc::new(BFTMemoryService::new()), 10);
         // Initialize the account.
         let account = Account::new(rng)?;
-        // Initialize the BFT.
-        let bft = BFT::new(account.clone(), storage.clone(), ledger.clone(), None, &[], None)?;
+
+        // Set up the BFT logic.
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
+        assert!(bft.is_timer_expired()); // 0 + 5 < now()
+
         // Set the leader certificate.
         let leader_certificate = sample_batch_certificate_for_round(2, rng);
         *bft.leader_certificate.write() = Some(leader_certificate);
@@ -1074,7 +1095,7 @@ mod tests {
         assert!(result);
 
         // Initialize a new BFT.
-        let bft_timer = BFT::new(account.clone(), storage.clone(), ledger.clone(), None, &[], None)?;
+        let bft_timer = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
         // If the leader certificate is not set and the timer has not expired, we are not ready for the next round.
         let result = bft_timer.is_even_round_ready_for_next_round(certificates.clone(), committee.clone(), 2);
         if !bft_timer.is_timer_expired() {
@@ -1105,7 +1126,8 @@ mod tests {
         assert_eq!(storage.max_gc_rounds(), 10);
 
         // Initialize the BFT.
-        let bft = BFT::new(account, storage, ledger, None, &[], None)?;
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
+        assert!(bft.is_timer_expired()); // 0 + 5 < now()
 
         // Ensure this call fails on an odd round.
         let result = bft.update_leader_certificate_to_even_round(1);
@@ -1123,7 +1145,8 @@ mod tests {
         assert_eq!(storage.max_gc_rounds(), 10);
 
         // Initialize the BFT.
-        let bft = BFT::new(account, storage, ledger, None, &[], None)?;
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
+        assert!(bft.is_timer_expired()); // 0 + 5 < now()
 
         // Ensure this call succeeds on an even round.
         let result = bft.update_leader_certificate_to_even_round(6);
@@ -1175,7 +1198,8 @@ mod tests {
 
         // Initialize the BFT.
         let account = Account::new(rng)?;
-        let bft = BFT::new(account, storage.clone(), ledger, None, &[], None)?;
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
+        assert!(bft.is_timer_expired()); // 0 + 5 < now()
 
         // Set the leader certificate.
         *bft.leader_certificate.write() = Some(leader_certificate);
@@ -1213,7 +1237,7 @@ mod tests {
             // Initialize the storage.
             let storage = Storage::new(ledger.clone(), Arc::new(BFTMemoryService::new()), 1);
             // Initialize the BFT.
-            let bft = BFT::new(account.clone(), storage, ledger.clone(), None, &[], None)?;
+            let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
 
             // Insert a mock DAG in the BFT.
             *bft.dag.write() = crate::helpers::dag::test_helpers::mock_dag_with_modified_last_committed_round(3);
@@ -1243,7 +1267,7 @@ mod tests {
             // Initialize the storage.
             let storage = Storage::new(ledger.clone(), Arc::new(BFTMemoryService::new()), 1);
             // Initialize the BFT.
-            let bft = BFT::new(account, storage, ledger, None, &[], None)?;
+            let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
 
             // Insert a mock DAG in the BFT.
             *bft.dag.write() = crate::helpers::dag::test_helpers::mock_dag_with_modified_last_committed_round(2);
@@ -1301,7 +1325,7 @@ mod tests {
         /* Test missing previous certificate. */
 
         // Initialize the BFT.
-        let bft = BFT::new(account, storage, ledger, None, &[], None)?;
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
 
         // The expected error message.
         let error_msg = format!(
@@ -1362,7 +1386,8 @@ mod tests {
 
         // Initialize the BFT.
         let account = Account::new(rng)?;
-        let bft = BFT::new(account, storage.clone(), ledger, None, &[], None)?;
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
+
         // Insert a mock DAG in the BFT.
         *bft.dag.write() = crate::helpers::dag::test_helpers::mock_dag_with_modified_last_committed_round(commit_round);
 
@@ -1428,7 +1453,7 @@ mod tests {
 
         // Initialize the BFT.
         let account = Account::new(rng)?;
-        let bft = BFT::new(account.clone(), storage, ledger.clone(), None, &[], None)?;
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
 
         // Insert a mock DAG in the BFT.
         *bft.dag.write() = crate::helpers::dag::test_helpers::mock_dag_with_modified_last_committed_round(commit_round);
@@ -1446,7 +1471,7 @@ mod tests {
         // Initialize a new instance of storage.
         let storage_2 = Storage::new(ledger.clone(), Arc::new(BFTMemoryService::new()), max_gc_rounds);
         // Initialize a new instance of BFT.
-        let bootup_bft = BFT::new(account, storage_2, ledger, None, &[], None)?;
+        let bootup_bft = initialize_bft(account.clone(), storage_2, ledger)?;
 
         // Sync the BFT DAG at bootup.
         bootup_bft.sync_bft_dag_at_bootup(certificates.clone()).await;
@@ -1600,7 +1625,7 @@ mod tests {
 
         // Initialize the BFT without bootup.
         let account = Account::new(rng)?;
-        let bft = BFT::new(account.clone(), storage, ledger.clone(), None, &[], None)?;
+        let bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
 
         // Insert a mock DAG in the BFT without bootup.
         *bft.dag.write() = crate::helpers::dag::test_helpers::mock_dag_with_modified_last_committed_round(0);
@@ -1625,7 +1650,7 @@ mod tests {
         let bootup_storage = Storage::new(ledger.clone(), Arc::new(BFTMemoryService::new()), max_gc_rounds);
 
         // Initialize a new instance of BFT with bootup.
-        let bootup_bft = BFT::new(account, bootup_storage.clone(), ledger.clone(), None, &[], None)?;
+        let bootup_bft = initialize_bft(account.clone(), bootup_storage.clone(), ledger.clone())?;
 
         // Sync the BFT DAG at bootup.
         bootup_bft.sync_bft_dag_at_bootup(pre_shutdown_certificates.clone()).await;
@@ -1803,7 +1828,7 @@ mod tests {
         }
         // Initialize the bootup BFT.
         let account = Account::new(rng)?;
-        let bootup_bft = BFT::new(account.clone(), storage.clone(), ledger.clone(), None, &[], None)?;
+        let bootup_bft = initialize_bft(account.clone(), storage.clone(), ledger.clone())?;
         // Insert a mock DAG in the BFT without bootup.
         *bootup_bft.dag.write() = crate::helpers::dag::test_helpers::mock_dag_with_modified_last_committed_round(0);
         // Sync the BFT DAG at bootup.

node/bft/src/gateway.rs

@@ -105,6 +105,8 @@ pub trait Transport<N: Network>: Send + Sync {
     fn broadcast(&self, event: Event<N>);
 }
 
+/// The gateway maintains connections to other validators.
+/// For connections with clients and provers, the Router logic is used.
 #[derive(Clone)]
 pub struct Gateway<N: Network> {
     /// The account of the node.
@@ -498,9 +500,8 @@ impl<N: Network> Gateway<N> {
         self.update_metrics();
     }
 
-    /// Inserts the given peer into the connected peers.
+    /// Inserts the given peer into the connected peers. This is only used in testing.
     #[cfg(test)]
-    // For unit tests, we need to make this public so we can inject peers.
     pub fn insert_connected_peer(&self, peer_ip: SocketAddr, peer_addr: SocketAddr, address: Address<N>) {
         // Adds a bidirectional map between the listener address and (ambiguous) peer address.
         self.resolver.insert_peer(peer_ip, peer_addr, address);

node/bft/src/helpers/channels.rs

@@ -60,6 +60,7 @@ pub fn init_consensus_channels<N: Network>() -> (ConsensusSender<N>, ConsensusRe
     (sender, receiver)
 }
 
+/// "Interface" that enables, for example, sending data from storage to the the BFT logic.
 #[derive(Clone, Debug)]
 pub struct BFTSender<N: Network> {
     pub tx_primary_round: mpsc::Sender<(u64, oneshot::Sender<bool>)>,
@@ -100,6 +101,7 @@ impl<N: Network> BFTSender<N> {
     }
 }
 
+/// Receiving counterpart to `BFTSender`
 #[derive(Debug)]
 pub struct BFTReceiver<N: Network> {
     pub rx_primary_round: mpsc::Receiver<(u64, oneshot::Sender<bool>)>,
@@ -108,7 +110,7 @@ pub struct BFTReceiver<N: Network> {
     pub rx_sync_bft: mpsc::Receiver<(BatchCertificate<N>, oneshot::Sender<Result<()>>)>,
 }
 
-/// Initializes the BFT channels.
+/// Initializes the BFT channels, and returns the sending and receiving ends.
 pub fn init_bft_channels<N: Network>() -> (BFTSender<N>, BFTReceiver<N>) {
     let (tx_primary_round, rx_primary_round) = mpsc::channel(MAX_CHANNEL_SIZE);
     let (tx_primary_certificate, rx_primary_certificate) = mpsc::channel(MAX_CHANNEL_SIZE);

node/bft/src/helpers/dag.rs

@@ -22,9 +22,12 @@ use snarkvm::{
 use indexmap::IndexSet;
 use std::collections::{BTreeMap, HashMap};
 
+/// Maintains an acyclic directed graph (DAG) of batches, from which we build a totally-ordered blockchain.
+/// The DAG is updated in rounds, where each validator adds at most one new batch.
 #[derive(Debug)]
 pub struct DAG<N: Network> {
     /// The in-memory collection of certificates that comprise the DAG.
+    /// For each round, there is a mapping from node address to batch.
     graph: BTreeMap<u64, HashMap<Address<N>, BatchCertificate<N>>>,
     /// The in-memory collection of recently committed certificate IDs (up to GC).
     recent_committed_ids: BTreeMap<u64, IndexSet<Field<N>>>,

node/bft/src/lib.rs

@@ -61,10 +61,10 @@ pub const MAX_TIMESTAMP_DELTA_IN_SECS: i64 = 10; // seconds
 /// The maximum number of workers that can be spawned.
 pub const MAX_WORKERS: u8 = 1; // worker(s)
 
-/// The frequency at which each primary broadcasts a ping to every other node.
+/// The interval at which each primary broadcasts a ping to every other node.
 /// Note: If this is updated, be sure to update `MAX_BLOCKS_BEHIND` to correspond properly.
 pub const PRIMARY_PING_IN_MS: u64 = 2 * MAX_BATCH_DELAY_IN_MS; // ms
-/// The frequency at which each worker broadcasts a ping to every other node.
+/// The interval at which each worker broadcasts a ping to every other node.
 pub const WORKER_PING_IN_MS: u64 = 4 * MAX_BATCH_DELAY_IN_MS; // ms
 
 /// A helper macro to spawn a blocking task.