apollo_network: using propeller in network manager

Author: sirandreww-starkware

Cargo.lock

@@ -64,6 +64,7 @@ members = [
   "crates/apollo_p2p_sync_config",
   "crates/apollo_proc_macros",
   "crates/apollo_proc_macros_tests",
+  "crates/apollo_propeller",
   "crates/apollo_protobuf",
   "crates/apollo_reverts",
   "crates/apollo_rpc",
@@ -179,6 +180,7 @@ apollo_p2p_sync.path = "crates/apollo_p2p_sync"
 apollo_p2p_sync_config.path = "crates/apollo_p2p_sync_config"
 apollo_proc_macros = { path = "crates/apollo_proc_macros", version = "0.0.0" }
 apollo_proc_macros_tests.path = "crates/apollo_proc_macros_tests"
+apollo_propeller.path = "crates/apollo_propeller"
 apollo_protobuf.path = "crates/apollo_protobuf"
 apollo_reverts.path = "crates/apollo_reverts"
 apollo_rpc.path = "crates/apollo_rpc"
@@ -209,6 +211,7 @@ assert_matches = "1.5.0"
 async-recursion = "1.1.0"
 async-stream = "0.3.3"
 async-trait = "0.1.79"
+asynchronous-codec = "0.7.0"
 atomic_refcell = "0.1.13"
 axum = "0.6.12"
 base64 = "0.13.0"
@@ -279,6 +282,7 @@ libp2p = "0.56.0"
 libp2p-swarm-test = "0.6.0"
 log = "0.4"
 lru = "0.12.0"
+lru_time_cache = "0.11.11"
 memmap2 = "0.8.0"
 mempool_test_utils.path = "crates/mempool_test_utils"
 metrics = "0.24.1"
@@ -315,10 +319,14 @@ prost-types = "0.12.1"
 protoc-prebuilt = "0.3.0"
 pyo3 = "0.19.1"
 pyo3-log = "0.8.1"
+quick-protobuf = "0.8.1"
+quick-protobuf-codec = "0.3.1"
+quickcheck = "1.0.3"
 quote = "1.0.26"
 rand = "0.8.5"
 rand_chacha = "0.3.1"
 rand_distr = "0.4.3"
+reed-solomon-simd = "3.1.0"
 regex = "1.10.4"
 replace_with = "0.1.7"
 reqwest = "0.12"

Cargo.toml

@@ -12,6 +12,7 @@ testing = []
 apollo_config.workspace = true
 apollo_metrics.workspace = true
 apollo_network_types.workspace = true
+apollo_propeller.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
 bytes.workspace = true

crates/apollo_network/Cargo.toml

@@ -3,28 +3,32 @@ use std::time::Duration;
 use futures::{FutureExt, StreamExt};
 use libp2p::core::multiaddr::Protocol;
 use libp2p::swarm::SwarmEvent;
-use libp2p::{Multiaddr, Swarm};
+use libp2p::{Multiaddr, PeerId, Swarm};
 use libp2p_swarm_test::SwarmExt;
 use starknet_api::core::ChainId;
 
 use crate::discovery::DiscoveryConfig;
 use crate::gossipsub_impl::Topic;
 use crate::mixed_behaviour::MixedBehaviour;
-use crate::network_manager::{BroadcastTopicClientTrait, GenericNetworkManager};
+use crate::network_manager::{
+    BroadcastTopicClientTrait,
+    GenericNetworkManager,
+    PropellerClientTrait,
+};
 use crate::peer_manager::PeerManagerConfig;
 use crate::{sqmr, Bytes};
 
 const TIMEOUT: Duration = Duration::from_secs(5);
 
-async fn create_swarm(bootstrap_peer_multiaddr: Option<Multiaddr>) -> Swarm<MixedBehaviour> {
+async fn create_swarm(bootstrap_peer_multiaddr: Option<Vec<Multiaddr>>) -> Swarm<MixedBehaviour> {
     let mut swarm = Swarm::new_ephemeral_tokio(|keypair| {
         MixedBehaviour::new(
             sqmr::Config::default(),
             DiscoveryConfig::default(),
             PeerManagerConfig::default(),
             None,
             keypair.clone(),
-            bootstrap_peer_multiaddr.map(|multiaddr| vec![multiaddr]),
+            bootstrap_peer_multiaddr,
             ChainId::Mainnet,
             None,
         )
@@ -60,6 +64,28 @@ fn create_network_manager(
     )
 }
 
+async fn create_network_managers(
+    num: usize,
+) -> Vec<(PeerId, GenericNetworkManager<Swarm<MixedBehaviour>>)> {
+    let mut bootstrap_addresses = vec![];
+    let mut network_managers = vec![];
+    for _ in 0..num {
+        let swarm = create_swarm(if bootstrap_addresses.is_empty() {
+            None
+        } else {
+            Some(bootstrap_addresses.clone())
+        })
+        .await;
+        let local_peer_id = swarm.local_peer_id();
+        let address = swarm.external_addresses().next().unwrap().clone();
+        // Add peer ID to the multiaddr for bootstrap addresses
+        let bootstrap_address = address.with_p2p(*local_peer_id).unwrap();
+        bootstrap_addresses.push(bootstrap_address);
+        network_managers.push((*local_peer_id, create_network_manager(swarm)));
+    }
+    network_managers
+}
+
 const BUFFER_SIZE: usize = 100;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -92,9 +118,9 @@ async fn broadcast_subscriber_end_to_end_test() {
         bootstrap_peer_multiaddr.with_p2p(*bootstrap_swarm.local_peer_id()).unwrap();
     let bootstrap_network_manager = create_network_manager(bootstrap_swarm);
     let mut network_manager1 =
-        create_network_manager(create_swarm(Some(bootstrap_peer_multiaddr.clone())).await);
+        create_network_manager(create_swarm(Some(vec![bootstrap_peer_multiaddr.clone()])).await);
     let mut network_manager2 =
-        create_network_manager(create_swarm(Some(bootstrap_peer_multiaddr)).await);
+        create_network_manager(create_swarm(Some(vec![bootstrap_peer_multiaddr])).await);
 
     let mut subscriber_channels1_1 =
         network_manager1.register_broadcast_topic::<Number>(topic1.clone(), BUFFER_SIZE).unwrap();
@@ -137,3 +163,47 @@ async fn broadcast_subscriber_end_to_end_test() {
         }
     }
 }
+
+#[tokio::test]
+async fn propeller_message_forwarding_test() {
+    // This test verifies that propeller messages are properly forwarded through channels
+    // when received by the network manager, similar to how gossipsub messages work.
+
+    let mut network_managers = create_network_managers(2).await;
+    let (peer_1, mut nm_1) = network_managers.remove(0);
+    let (peer_2, mut nm_2) = network_managers.remove(0);
+
+    let peers = vec![(peer_1, 1000), (peer_2, 500)];
+
+    let mut channels_1 =
+        nm_1.register_propeller_channels::<Vec<u8>>(BUFFER_SIZE, peers.clone()).unwrap();
+    let mut channels_2 = nm_2.register_propeller_channels::<Vec<u8>>(BUFFER_SIZE, peers).unwrap();
+
+    let message = vec![123; 64]; // must be a multiple of 64
+    let message_id = 1;
+
+    tokio::select! {
+        _ = nm_1.run() => panic!("network manager ended"),
+        _ = nm_2.run() => panic!("network manager ended"),
+        result = tokio::time::timeout(
+            TIMEOUT, async move {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+
+                println!("Sending message");
+                channels_1.propeller_client.send_message(message.clone(), message_id).await.unwrap();
+
+                println!("Receiving message");
+                let (received_message_id, received_message) =
+                    channels_2.propeller_messages_receiver.next().await.unwrap();
+
+                assert_eq!(received_message.unwrap(), message);
+                assert_eq!(received_message_id, message_id);
+
+                assert!(channels_1.propeller_messages_receiver.next().now_or_never().is_none());
+                assert!(channels_2.propeller_messages_receiver.next().now_or_never().is_none());
+            }
+        ) => {
+            result.unwrap()
+        }
+    }
+}

crates/apollo_network/src/e2e_broadcast_test.rs

@@ -13,6 +13,7 @@ pub mod misconduct_score;
 mod mixed_behaviour;
 pub mod network_manager;
 mod peer_manager;
+pub mod propeller_impl;
 mod sqmr;
 #[cfg(test)]
 mod test_utils;