apollo_network: propeller

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

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

crates/apollo_consensus_manager/src/consensus_manager.rs

@@ -57,6 +57,7 @@ pub fn create_p2p_propagator_and_runner(
         broadcast_metrics_by_topic: Some(broadcast_metrics_by_topic),
         sqmr_metrics: None,
         event_metrics: Some(EventMetrics { event_counter: MEMPOOL_P2P_NETWORK_EVENTS }),
+        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,9 @@ pub enum NetworkProtocol {
     /// Use Reversed SQMR where receivers initiate requests to broadcasters
     #[value(name = "reversed-sqmr")]
     ReveresedSqmr,
+    /// Use Propeller for leader-based erasure-coded broadcasting
+    #[value(name = "propeller")]
+    Propeller,
 }
 
 impl Display for Mode {

crates/apollo_network/src/bin/broadcast_network_stress_test_node/args.rs

@@ -2,6 +2,10 @@ use apollo_network::network_manager::{
     BroadcastTopicClient,
     BroadcastTopicClientTrait,
     BroadcastTopicServer,
+    PropellerClient,
+    PropellerClientTrait,
+    PropellerMessageServer,
+    ReceivedPropellerMessage,
     SqmrClientSender,
     SqmrServerReceiver,
 };
@@ -16,6 +20,26 @@ pub enum MessageSender {
     Gossipsub(BroadcastTopicClient<TopicType>),
     Sqmr(SqmrClientSender<TopicType, TopicType>),
     ReveresedSqmr(ReveresedSqmrSender),
+    Propeller(PropellerSender),
+}
+
+/// Wrapper for Propeller client that handles message ID generation
+pub struct PropellerSender {
+    client: PropellerClient<TopicType>,
+}
+
+impl PropellerSender {
+    pub fn new(client: PropellerClient<TopicType>) -> Self {
+        Self { client }
+    }
+
+    async fn send_message(&mut self, message: TopicType) {
+        if let Err(e) = self.client.send_message(message).await {
+            error!("Failed to send Propeller message: {:?}", e);
+        } else {
+            trace!("Sent Propeller message");
+        }
+    }
 }
 
 /// Wrapper for ReveresedSqmr that maintains the last active query
@@ -88,6 +112,9 @@ impl MessageSender {
                 // Then broadcast the message to all active queries
                 sender.broadcast_to_queries(message).await;
             }
+            MessageSender::Propeller(sender) => {
+                sender.send_message(message).await;
+            }
         }
     }
 }
@@ -96,6 +123,7 @@ pub enum MessageReceiver {
     Gossipsub(BroadcastTopicServer<TopicType>),
     Sqmr(SqmrServerReceiver<TopicType, TopicType>),
     ReveresedSqmr(SqmrClientSender<TopicType, TopicType>),
+    Propeller(PropellerMessageServer<TopicType>),
 }
 
 impl MessageReceiver {
@@ -146,6 +174,24 @@ impl MessageReceiver {
                     }
                 }
             },
+            MessageReceiver::Propeller(receiver) => receiver
+                .for_each(
+                    |(_, message_root, result): ReceivedPropellerMessage<TopicType>| async move {
+                        match result {
+                            Ok(message) => {
+                                trace!("Received Propeller message with ID: {}", message_root);
+                                f(message);
+                            }
+                            Err(e) => {
+                                error!(
+                                    "Failed to deserialize Propeller message {}: {:?}",
+                                    message_root, e
+                                );
+                            }
+                        }
+                    },
+                )
+                .await,
         }
     }
 }