apollo_network: propeller

Author: sirandreww-starkware

Cargo.lock

@@ -67,6 +67,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",
@@ -184,6 +185,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"
@@ -216,6 +218,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"
@@ -286,6 +289,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"
@@ -320,10 +324,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

@@ -174,6 +174,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

@@ -60,6 +60,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,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 {
@@ -147,6 +175,24 @@ impl MessageReceiver {
                     }
                 }
             },
+            MessageReceiver::Propeller(receiver) => receiver
+                .for_each(
+                    |(sender, message_root, result): ReceivedPropellerMessage<TopicType>| async move {
+                        match result {
+                            Ok(message) => {
+                                trace!("Received Propeller message with ID: {}", message_root);
+                                f(message, Some(sender));
+                            }
+                            Err(e) => {
+                                error!(
+                                    "Failed to deserialize Propeller message {}: {:?}",
+                                    message_root, e
+                                );
+                            }
+                        }
+                    },
+                )
+                .await,
         }
     }
 }

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

@@ -12,6 +12,12 @@ use apollo_network::network_manager::metrics::{
     EVENT_TYPE_LABELS,
     NETWORK_BROADCAST_DROP_LABELS,
 };
+use apollo_propeller::metrics::{
+    PropellerMetrics,
+    COLLECTION_LENGTH_LABELS,
+    SHARD_SEND_FAILURE_LABELS,
+    SHARD_VALIDATION_FAILURE_LABELS,
+};
 use libp2p::gossipsub::{Sha256Topic, Topic};
 use libp2p::PeerId;
 use sysinfo::{Networks, System};
@@ -63,6 +69,24 @@ define_metrics!(
         MetricCounter { NETWORK_RESET_TOTAL, "network_reset_total", "Total number of network resets performed", init = 0 },
         LabeledMetricCounter { NETWORK_DROPPED_BROADCAST_MESSAGES, "network_dropped_broadcast_messages", "Number of dropped broadcast messages by reason", init = 0, labels = NETWORK_BROADCAST_DROP_LABELS },
         LabeledMetricCounter { NETWORK_EVENT_COUNTER, "network_event_counter", "Network events counter by type", init = 0, labels = EVENT_TYPE_LABELS },
+
+        // Propeller Protocol Metrics
+        MetricCounter { PROPELLER_SHARDS_PUBLISHED, "propeller_shards_published", "Total number of shards published (created) by this node", init = 0 },
+        MetricCounter { PROPELLER_SHARDS_SENT, "propeller_shards_sent", "Total number of shards sent to peers (includes forwarding)", init = 0 },
+        LabeledMetricCounter { PROPELLER_SHARDS_SEND_FAILED, "propeller_shards_send_failed", "Total number of shard send failures, labeled by reason", init = 0, labels = SHARD_SEND_FAILURE_LABELS },
+        MetricCounter { PROPELLER_SHARD_BYTES_SENT, "propeller_shard_bytes_sent", "Total bytes sent in shard data (payload only)", init = 0 },
+        MetricCounter { PROPELLER_SHARDS_RECEIVED, "propeller_shards_received", "Total number of shards received from peers", init = 0 },
+        MetricCounter { PROPELLER_SHARDS_VALIDATED, "propeller_shards_validated", "Total number of shards successfully validated", init = 0 },
+        LabeledMetricCounter { PROPELLER_SHARDS_VALIDATION_FAILED, "propeller_shards_validation_failed", "Total number of shards that failed validation, labeled by reason", init = 0, labels = SHARD_VALIDATION_FAILURE_LABELS },
+        MetricCounter { PROPELLER_SHARDS_FORWARDED, "propeller_shards_forwarded", "Total number of shards forwarded to children in tree", init = 0 },
+        MetricCounter { PROPELLER_SHARD_BYTES_RECEIVED, "propeller_shard_bytes_received", "Total bytes received in shard data (payload only)", init = 0 },
+        MetricCounter { PROPELLER_MESSAGES_RECONSTRUCTED, "propeller_messages_reconstructed", "Total number of messages successfully reconstructed from shards", init = 0 },
+        MetricCounter { PROPELLER_MESSAGES_RECONSTRUCTION_FAILED, "propeller_messages_reconstruction_failed", "Total number of message reconstruction failures", init = 0 },
+        MetricCounter { PROPELLER_TREES_GENERATED, "propeller_trees_generated", "Total number of tree generation operations", init = 0 },
+        LabeledMetricGauge { PROPELLER_COLLECTION_LENGTHS, "propeller_collection_lengths", "Length of various collections (queues, sets, caches) tracked by label", labels = COLLECTION_LENGTH_LABELS },
+        MetricHistogram { PROPELLER_SHARD_VALIDATION_DURATION, "propeller_shard_validation_duration", "Time to validate a single shard (seconds)" },
+        MetricHistogram { PROPELLER_MESSAGE_RECONSTRUCTION_DURATION, "propeller_message_reconstruction_duration", "Time to reconstruct a message from shards (seconds)" },
+        MetricHistogram { PROPELLER_MESSAGE_END_TO_END_LATENCY, "propeller_message_end_to_end_latency", "End-to-end latency from first shard received to message reconstructed (seconds)" },
     },
 );
 
@@ -150,12 +174,32 @@ pub fn create_network_metrics() -> NetworkMetrics {
     // Create event metrics for network events monitoring
     let event_metrics = EventMetrics { event_counter: NETWORK_EVENT_COUNTER };
 
+    let propeller_metrics = PropellerMetrics {
+        shards_published: PROPELLER_SHARDS_PUBLISHED,
+        shards_sent: PROPELLER_SHARDS_SENT,
+        shards_send_failed: PROPELLER_SHARDS_SEND_FAILED,
+        shard_bytes_sent: PROPELLER_SHARD_BYTES_SENT,
+        shards_received: PROPELLER_SHARDS_RECEIVED,
+        shards_validated: PROPELLER_SHARDS_VALIDATED,
+        shards_validation_failed: PROPELLER_SHARDS_VALIDATION_FAILED,
+        shards_forwarded: PROPELLER_SHARDS_FORWARDED,
+        shard_bytes_received: PROPELLER_SHARD_BYTES_RECEIVED,
+        messages_reconstructed: PROPELLER_MESSAGES_RECONSTRUCTED,
+        messages_reconstruction_failed: PROPELLER_MESSAGES_RECONSTRUCTION_FAILED,
+        trees_generated: PROPELLER_TREES_GENERATED,
+        collection_lengths: PROPELLER_COLLECTION_LENGTHS,
+        shard_validation_duration: PROPELLER_SHARD_VALIDATION_DURATION,
+        message_reconstruction_duration: PROPELLER_MESSAGE_RECONSTRUCTION_DURATION,
+        message_end_to_end_latency: PROPELLER_MESSAGE_END_TO_END_LATENCY,
+    };
+
     NetworkMetrics {
         num_connected_peers: NETWORK_CONNECTED_PEERS,
         num_blacklisted_peers: NETWORK_BLACKLISTED_PEERS,
         broadcast_metrics_by_topic: Some(broadcast_metrics_by_topic),
         sqmr_metrics: Some(sqmr_metrics),
         event_metrics: Some(event_metrics),
+        propeller_metrics: Some(propeller_metrics),
     }
 }