apollo_network: propeller

Author: sirandreww-starkware

Cargo.lock

@@ -69,6 +69,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",
@@ -187,6 +188,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"
@@ -219,6 +221,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"
@@ -290,6 +293,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"
@@ -323,10 +327,15 @@ 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"
+rayon = "1.11.0"
+reed-solomon-simd = "3.1.0"
 regex = "1.10.4"
 replace_with = "0.1.7"
 reqwest = "0.12"

Cargo.toml

@@ -214,6 +214,7 @@ impl ConsensusManager {
             sqmr_metrics: None,
             event_metrics: Some(EventMetrics { event_counter: CONSENSUS_NETWORK_EVENTS }),
             latency_metrics: Some(LatencyMetrics { ping_latency_seconds: CONSENSUS_PING_LATENCY }),
+            propeller_metrics: None,
         });
 
         NetworkManager::new(self.config.network_config.clone(), None, network_manager_metrics)

crates/apollo_consensus_manager/src/consensus_manager.rs

@@ -74,6 +74,7 @@ pub fn create_p2p_propagator_and_runner(
         sqmr_metrics: None,
         event_metrics: Some(EventMetrics { event_counter: MEMPOOL_P2P_NETWORK_EVENTS }),
         latency_metrics: Some(LatencyMetrics { ping_latency_seconds: MEMPOOL_P2P_PING_LATENCY }),
+        propeller_metrics: None,
     });
     let mut network_manager = NetworkManager::new(
         mempool_p2p_config.network_config,

crates/apollo_mempool_p2p/src/lib.rs

@@ -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

@@ -33,6 +33,7 @@ impl DiscoveryMixedBehaviour {
             PeerManagerConfig::default(),
             None, // No event tracker for tests
             None, // No latency metrics for tests
+            None,
             key,
             bootstrap_peer_multiaddr,
             ChainId::Mainnet,

crates/apollo_network/src/discovery/flow_test.rs

@@ -3,30 +3,35 @@ 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::prune_dead_connections::{DEFAULT_PING_INTERVAL, DEFAULT_PING_TIMEOUT};
 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,
             None,
+            None,
             keypair.clone(),
-            bootstrap_peer_multiaddr.map(|multiaddr| vec![multiaddr]),
+            bootstrap_peer_multiaddr,
             ChainId::Mainnet,
             None,
             DEFAULT_PING_INTERVAL,
@@ -64,6 +69,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)]
@@ -96,9 +123,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();
@@ -141,3 +168,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()).await.unwrap();
+    let mut channels_2 =
+        nm_2.register_propeller_channels::<Vec<u8>>(BUFFER_SIZE, peers).await.unwrap();
+
+    let message = vec![123; 64]; // must be a multiple of 64
+
+    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()).await.unwrap();
+
+                println!("Receiving message");
+                let (publisher, _, received_message) =
+                    channels_2.propeller_messages_receiver.next().await.unwrap();
+
+                assert_eq!(received_message.unwrap(), message);
+                assert_eq!(publisher, peer_1);
+
+                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

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

crates/apollo_network/src/lib.rs

@@ -8,6 +8,7 @@ use apollo_metrics::metrics::{
     MetricGauge,
     MetricHistogram,
 };
+use apollo_propeller::metrics::PropellerMetrics;
 use libp2p::gossipsub::{PublishError, TopicHash};
 use strum::{IntoStaticStr, VariantNames};
 use strum_macros::EnumVariantNames;
@@ -195,6 +196,7 @@ pub struct NetworkMetrics {
     pub sqmr_metrics: Option<SqmrNetworkMetrics>,
     pub event_metrics: Option<EventMetrics>,
     pub latency_metrics: Option<LatencyMetrics>,
+    pub propeller_metrics: Option<PropellerMetrics>,
 }
 
 impl NetworkMetrics {
@@ -217,5 +219,8 @@ impl NetworkMetrics {
         if let Some(latency_metrics) = self.latency_metrics.as_ref() {
             latency_metrics.register();
         }
+        if let Some(propeller_metrics) = self.propeller_metrics.as_ref() {
+            propeller_metrics.register();
+        }
     }
 }