misc(bft): replace PrimarySender with a callback

Author: kaimast

Cargo.lock

@@ -38,11 +38,15 @@ cuda = [
   "snarkos-node-bft-ledger-service/cuda",
   "snarkos-node-sync/cuda"
 ]
+persistent-storage = [ "snarkos-node-bft-storage-service/persistent" ]
 test = [
   # "snarkvm/test" this breaks some of the tests
-  "snarkvm/test-helpers",
   "snarkos-node-bft-ledger-service/test",
-  "snarkos-node-bft-storage-service/test"
+  "snarkos-node-bft-storage-service/test",
+  "test-helpers"
+]
+test-helpers = [
+  "snarkvm/test-helpers",
 ]
 serial = [ "snarkos-node-bft-ledger-service/serial" ]
 
@@ -206,3 +210,8 @@ workspace = true
 
 [dev-dependencies.mockall]
 version = "0.13"
+
+[[test]]
+name = "gateway-e2e"
+path = "./tests/gateway_e2e.rs"
+required-features = [ "test-helpers" ]

node/bft/Cargo.toml

@@ -20,13 +20,7 @@ extern crate tracing;
 extern crate snarkos_node_metrics as metrics;
 
 use snarkos_account::Account;
-use snarkos_node_bft::{
-    BFT,
-    BftCallback,
-    MEMORY_POOL_PORT,
-    Primary,
-    helpers::{PrimarySender, Storage, init_primary_channels},
-};
+use snarkos_node_bft::{BFT, BftCallback, MEMORY_POOL_PORT, Primary, helpers::Storage};
 use snarkos_node_bft_ledger_service::TranslucentLedgerService;
 use snarkos_node_bft_storage_service::BFTMemoryService;
 use snarkos_node_sync::BlockSync;
@@ -69,7 +63,7 @@ use std::{
     str::FromStr,
     sync::{Arc, Mutex, OnceLock},
 };
-use tokio::{net::TcpListener, sync::oneshot};
+use tokio::net::TcpListener;
 use tracing_subscriber::{
     layer::{Layer, SubscriberExt},
     util::SubscriberInitExt,
@@ -119,13 +113,7 @@ pub fn initialize_logger(verbosity: u8) {
 /**************************************************************************************************/
 
 /// Starts the BFT instance.
-pub async fn start_bft(
-    node_id: u16,
-    num_nodes: u16,
-    peers: HashMap<u16, SocketAddr>,
-) -> Result<(BFT<CurrentNetwork>, PrimarySender<CurrentNetwork>)> {
-    // Initialize the primary channels.
-    let (sender, receiver) = init_primary_channels();
+pub async fn start_bft(node_id: u16, num_nodes: u16, peers: HashMap<u16, SocketAddr>) -> Result<BFT<CurrentNetwork>> {
     // Initialize the components.
     let (committee, account) = initialize_components(node_id, num_nodes)?;
     // Initialize the translucent ledger service.
@@ -148,26 +136,25 @@ pub async fn start_bft(
     let consensus_handler = Arc::new(ConsensusHandler {});
     // Initialize the BFT instance.
     let block_sync = Arc::new(BlockSync::new(ledger.clone()));
-    let mut bft =
-        BFT::<CurrentNetwork>::new(account, storage, ledger, block_sync, ip, &trusted_validators, storage_mode, None)?;
+    let bft =
+        BFT::<CurrentNetwork>::new(account, storage, ledger, block_sync, ip, &trusted_validators, storage_mode, None)
+            .await?;
     // Run the BFT instance.
-    bft.run(None, Some(consensus_handler), sender.clone(), receiver).await?;
+    bft.run(None, Some(consensus_handler)).await?;
     // Retrieve the BFT's primary.
     let primary = bft.primary();
     // Handle OS signals.
     handle_signals(primary);
     // Return the BFT instance.
-    Ok((bft, sender))
+    Ok(bft)
 }
 
 /// Starts the primary instance.
 pub async fn start_primary(
     node_id: u16,
     num_nodes: u16,
     peers: HashMap<u16, SocketAddr>,
-) -> Result<(Primary<CurrentNetwork>, PrimarySender<CurrentNetwork>)> {
-    // Initialize the primary channels.
-    let (sender, receiver) = init_primary_channels();
+) -> Result<Primary<CurrentNetwork>> {
     // Initialize the components.
     let (committee, account) = initialize_components(node_id, num_nodes)?;
     // Initialize the translucent ledger service.
@@ -188,7 +175,7 @@ pub async fn start_primary(
     let trusted_validators = trusted_validators(node_id, num_nodes, peers);
     // Initialize the primary instance.
     let block_sync = Arc::new(BlockSync::new(ledger.clone()));
-    let mut primary = Primary::<CurrentNetwork>::new(
+    let primary = Primary::<CurrentNetwork>::new(
         account,
         storage,
         ledger,
@@ -197,13 +184,14 @@ pub async fn start_primary(
         &trusted_validators,
         storage_mode,
         None,
-    )?;
+    )
+    .await?;
     // Run the primary instance.
-    primary.run(None, None, None, sender.clone(), receiver).await?;
+    primary.run(None, None, None).await?;
     // Handle OS signals.
     handle_signals(&primary);
     // Return the primary instance.
-    Ok((primary, sender))
+    Ok(primary)
 }
 
 /// Initialize the translucent ledger service.
@@ -371,8 +359,7 @@ fn handle_signals(primary: &Primary<CurrentNetwork>) {
 /**************************************************************************************************/
 
 /// Fires *fake* unconfirmed solutions at the node.
-fn fire_unconfirmed_solutions(sender: &PrimarySender<CurrentNetwork>, node_id: u16, interval_ms: u64) {
-    let tx_unconfirmed_solution = sender.tx_unconfirmed_solution.clone();
+fn fire_unconfirmed_solutions(primary: Primary<CurrentNetwork>, node_id: u16, interval_ms: u64) {
     tokio::task::spawn(async move {
         // This RNG samples the *same* fake solutions for all nodes.
         let mut shared_rng = rand_chacha::ChaChaRng::seed_from_u64(123456789);
@@ -396,13 +383,8 @@ fn fire_unconfirmed_solutions(sender: &PrimarySender<CurrentNetwork>, node_id: u
             // Sample a random fake solution ID and solution.
             let (solution_id, solution) =
                 if counter % 2 == 0 { sample(&mut shared_rng) } else { sample(&mut unique_rng) };
-            // Initialize a callback sender and receiver.
-            let (callback, callback_receiver) = oneshot::channel();
             // Send the fake solution.
-            if let Err(e) = tx_unconfirmed_solution.send((solution_id, solution, callback)).await {
-                error!("Failed to send unconfirmed solution: {e}");
-            }
-            let _ = callback_receiver.await;
+            let _ = primary.process_unconfirmed_solution(solution_id, solution).await;
             // Increment the counter.
             counter += 1;
             // Sleep briefly.
@@ -412,8 +394,7 @@ fn fire_unconfirmed_solutions(sender: &PrimarySender<CurrentNetwork>, node_id: u
 }
 
 /// Fires *fake* unconfirmed transactions at the node.
-fn fire_unconfirmed_transactions(sender: &PrimarySender<CurrentNetwork>, node_id: u16, interval_ms: u64) {
-    let tx_unconfirmed_transaction = sender.tx_unconfirmed_transaction.clone();
+fn fire_unconfirmed_transactions(primary: Primary<CurrentNetwork>, node_id: u16, interval_ms: u64) {
     tokio::task::spawn(async move {
         // This RNG samples the *same* fake transactions for all nodes.
         let mut shared_rng = rand_chacha::ChaChaRng::seed_from_u64(123456789);
@@ -438,13 +419,8 @@ fn fire_unconfirmed_transactions(sender: &PrimarySender<CurrentNetwork>, node_id
         loop {
             // Sample a random fake transaction ID and transaction.
             let (id, transaction) = if counter % 2 == 0 { sample(&mut shared_rng) } else { sample(&mut unique_rng) };
-            // Initialize a callback sender and receiver.
-            let (callback, callback_receiver) = oneshot::channel();
             // Send the fake transaction.
-            if let Err(e) = tx_unconfirmed_transaction.send((id, transaction, callback)).await {
-                error!("Failed to send unconfirmed transaction: {e}");
-            }
-            let _ = callback_receiver.await;
+            let _ = primary.process_unconfirmed_transaction(id, transaction).await;
             // Increment the counter.
             counter += 1;
             // Sleep briefly.
@@ -583,14 +559,14 @@ async fn main() -> Result<()> {
     let mut bft_holder = None;
 
     // Start the node.
-    let (primary, sender) = match args.mode {
+    let primary = match args.mode {
         Mode::Bft => {
             // Start the BFT.
-            let (bft, sender) = start_bft(args.id, args.num_nodes, peers).await?;
+            let bft = start_bft(args.id, args.num_nodes, peers).await?;
             // Set the BFT holder.
             bft_holder = Some(bft.clone());
             // Return the primary and sender.
-            (bft.primary().clone(), sender)
+            bft.primary().clone()
         }
         Mode::Narwhal => start_primary(args.id, args.num_nodes, peers).await?,
     };
@@ -602,7 +578,7 @@ async fn main() -> Result<()> {
     match (args.fire_transmissions, args.fire_solutions) {
         // Note: We allow the user to overload the solutions rate, even when the 'fire-transmissions' flag is enabled.
         (Some(rate), _) | (_, Some(rate)) => {
-            fire_unconfirmed_solutions(&sender, args.id, rate.unwrap_or(DEFAULT_INTERVAL_MS));
+            fire_unconfirmed_solutions(primary.clone(), args.id, rate.unwrap_or(DEFAULT_INTERVAL_MS));
         }
         _ => (),
     };
@@ -611,7 +587,7 @@ async fn main() -> Result<()> {
     match (args.fire_transmissions, args.fire_transactions) {
         // Note: We allow the user to overload the transactions rate, even when the 'fire-transmissions' flag is enabled.
         (Some(rate), _) | (_, Some(rate)) => {
-            fire_unconfirmed_transactions(&sender, args.id, rate.unwrap_or(DEFAULT_INTERVAL_MS));
+            fire_unconfirmed_transactions(primary.clone(), args.id, rate.unwrap_or(DEFAULT_INTERVAL_MS));
         }
         _ => (),
     };

node/bft/examples/simple_node.rs

@@ -17,13 +17,17 @@ use snarkvm::{
     ledger::{
         block::{Block, Transaction},
         committee::Committee,
-        narwhal::{BatchCertificate, Data, Subdag, Transmission, TransmissionID},
+        narwhal::{BatchCertificate, Data, Transmission, TransmissionID},
         puzzle::{Solution, SolutionID},
     },
     prelude::{Address, ConsensusVersion, Field, Network, Result},
 };
 
+#[cfg(feature = "ledger-write")]
 use indexmap::IndexMap;
+#[cfg(feature = "ledger-write")]
+use snarkvm::ledger::narwhal::Subdag;
+
 use std::{fmt::Debug, ops::Range};
 
 #[async_trait]