Subscribe to topic and decode SignedSSVMessage

anchor/network/Cargo.toml

@@ -9,7 +9,9 @@ dirs = { workspace = true }
 discv5 = { workspace = true }
 ethereum_serde_utils = "1.0.0-beta.0"
 ethereum_ssz = "0.8.1"
+ethereum_ssz_derive = "0.8.1"
 futures = { workspace = true }
+hex = "0.4.3"
 libp2p = { version = "0.54", default-features = false, features = [
   "identify",
   "yamux",

anchor/network/src/network.rs

@@ -5,15 +5,15 @@ use std::time::Duration;
 use futures::StreamExt;
 use libp2p::core::muxing::StreamMuxerBox;
 use libp2p::core::transport::Boxed;
-use libp2p::gossipsub::{MessageAuthenticity, ValidationMode};
+use libp2p::gossipsub::{IdentTopic, MessageAuthenticity, ValidationMode};
 use libp2p::identity::Keypair;
 use libp2p::multiaddr::Protocol;
 use libp2p::swarm::SwarmEvent;
 use libp2p::{futures, gossipsub, identify, ping, PeerId, Swarm, SwarmBuilder};
 use lighthouse_network::discovery::DiscoveredPeers;
 use lighthouse_network::discv5::enr::k256::sha2::{Digest, Sha256};
 use task_executor::TaskExecutor;
-use tracing::{info, log};
+use tracing::{debug, info, log, warn};
 
 use crate::behaviour::AnchorBehaviour;
 use crate::behaviour::AnchorBehaviourEvent;
@@ -22,7 +22,9 @@ use crate::keypair_utils::load_private_key;
 use crate::transport::build_transport;
 use crate::Config;
 
+use crate::types::ssv_message::SignedSSVMessage;
 use lighthouse_network::EnrExt;
+use ssz::Decode;
 
 pub struct Network {
     swarm: Swarm<AnchorBehaviour>,
@@ -83,12 +85,49 @@ impl Network {
 
     /// Main loop for polling and handling swarm and channels.
     pub async fn run(mut self) {
+        let topic = IdentTopic::new("ssv.v2.9");
+
+        match self.swarm.behaviour_mut().gossipsub.subscribe(&topic) {
+            Err(e) => {
+                warn!(topic = %topic, "error" = ?e, "Failed to subscribe to topic");
+            }
+            Ok(_) => {
+                debug!(topic = %topic, "Subscribed to topic");
+            }
+        }
+
         loop {
             tokio::select! {
                 swarm_message = self.swarm.select_next_some() => {
                     match swarm_message {
                         SwarmEvent::Behaviour(behaviour_event) => match behaviour_event {
-                            AnchorBehaviourEvent::Gossipsub(_ge) => {
+                            AnchorBehaviourEvent::Gossipsub(ge) => {
+                                match ge {
+                                    gossipsub::Event::Message {
+                                        propagation_source,
+                                        message_id,
+                                        message,
+                                    } => {
+                                        log::debug!(
+                                            "Received message from {:?} with id {:?}: {:?}",
+                                            propagation_source,
+                                            message_id,
+                                            &message
+                                        );
+                                        match SignedSSVMessage::from_ssz_bytes(&message.data) {
+                                            Ok(deserialized_message) => {
+                                                log::debug!("SSVMessage: {:?}", deserialized_message);
+                                            }
+                                            Err(e) => {
+                                                log::error!("Error deserializing SSVMessage: {:?}", e);
+                                            }
+                                        }
+                                    }
+                                    // TODO handle gossipsub events
+                                    _ => {
+                                        log::debug!("Unhandled gossipsub event: {:?}", ge);
+                                    }
+                                }
                                 // TODO handle gossipsub events
                             },
                             // Inform the peer manager about discovered peers.
@@ -151,7 +190,7 @@ async fn build_anchor_behaviour(
         .duplicate_cache_time(duplicate_cache_time)
         .message_id_fn(gossip_message_id)
         .flood_publish(false)
-        .validation_mode(ValidationMode::Strict)
+        .validation_mode(ValidationMode::Permissive)
         .mesh_n(8) //D
         .mesh_n_low(6) // Dlo
         .mesh_n_high(12) // Dhi

anchor/network/src/types.rs

@@ -1 +1,2 @@
 mod gossip_kind;
+pub mod ssv_message;