Added error handling to pubsub

Author: 0xMimir

node/src/action_kind.rs

@@ -310,8 +310,10 @@ pub enum ActionKind {
     P2pNetworkPubsubNewStream,
     P2pNetworkPubsubOutgoingData,
     P2pNetworkPubsubOutgoingMessage,
+    P2pNetworkPubsubOutgoingMessageError,
     P2pNetworkPubsubPrune,
     P2pNetworkPubsubSign,
+    P2pNetworkPubsubSignError,
     P2pNetworkRpcHeartbeatSend,
     P2pNetworkRpcIncomingData,
     P2pNetworkRpcIncomingMessage,
@@ -538,7 +540,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 446;
+    pub const COUNT: u16 = 448;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1466,8 +1468,10 @@ impl ActionKindGet for P2pNetworkPubsubAction {
             Self::Prune { .. } => ActionKind::P2pNetworkPubsubPrune,
             Self::Broadcast { .. } => ActionKind::P2pNetworkPubsubBroadcast,
             Self::Sign { .. } => ActionKind::P2pNetworkPubsubSign,
+            Self::SignError { .. } => ActionKind::P2pNetworkPubsubSignError,
             Self::BroadcastSigned { .. } => ActionKind::P2pNetworkPubsubBroadcastSigned,
             Self::OutgoingMessage { .. } => ActionKind::P2pNetworkPubsubOutgoingMessage,
+            Self::OutgoingMessageError { .. } => ActionKind::P2pNetworkPubsubOutgoingMessageError,
             Self::OutgoingData { .. } => ActionKind::P2pNetworkPubsubOutgoingData,
         }
     }

p2p/src/network/pubsub/p2p_network_pubsub_actions.rs

@@ -37,13 +37,23 @@ pub enum P2pNetworkPubsubAction {
         data: Data,
         topic: String,
     },
+    #[action_event(level = warn, fields(display(author), display(topic)))]
+    SignError {
+        author: PeerId,
+        topic: String,
+    },
     BroadcastSigned {
         signature: Data,
     },
     OutgoingMessage {
         msg: pb::Rpc,
         peer_id: PeerId,
     },
+    #[action_event(level = warn, fields(display(peer_id), debug(msg)))]
+    OutgoingMessageError {
+        msg: pb::Rpc,
+        peer_id: PeerId,
+    },
     OutgoingData {
         data: Data,
         peer_id: PeerId,

p2p/src/network/pubsub/p2p_network_pubsub_effects.rs

@@ -100,7 +100,9 @@ impl P2pNetworkPubsubAction {
             P2pNetworkPubsubAction::Broadcast { message } => {
                 // println!("(pubsub) {this} broadcast");
                 let mut buffer = vec![0; 8];
-                binprot::BinProtWrite::binprot_write(&message, &mut buffer).expect("msg");
+                if binprot::BinProtWrite::binprot_write(&message, &mut buffer).is_err() {
+                    return;
+                }
                 let len = buffer.len() - 8;
                 buffer[..8].clone_from_slice(&(len as u64).to_le_bytes());
 
@@ -111,12 +113,11 @@ impl P2pNetworkPubsubAction {
                     topic: TOPIC.to_owned(),
                 });
             }
-            P2pNetworkPubsubAction::Sign { .. } => {
+            P2pNetworkPubsubAction::Sign { author, topic, .. } => {
                 if let Some(to_sign) = state.to_sign.front() {
                     let mut publication = vec![];
-                    if let Err(err) = prost::Message::encode(to_sign, &mut publication) {
-                        // TODO: dispatch action for logging
-                        let _ = err;
+                    if prost::Message::encode(to_sign, &mut publication).is_err() {
+                        store.dispatch(P2pNetworkPubsubAction::SignError { author, topic });
                     } else {
                         let signature = store.service().sign_publication(&publication).into();
                         store.dispatch(P2pNetworkPubsubAction::BroadcastSigned { signature });
@@ -183,17 +184,20 @@ impl P2pNetworkPubsubAction {
                     //     println!("{}", std::str::from_utf8(&id).unwrap());
                     // }
                     let mut data = vec![];
-                    if let Err(err) = prost::Message::encode_length_delimited(&msg, &mut data) {
-                        // TODO: dispatch action for logging
-                        let _ = err;
+                    if prost::Message::encode_length_delimited(&msg, &mut data).is_err() {
+                        store.dispatch(P2pNetworkPubsubAction::OutgoingMessageError {
+                            msg,
+                            peer_id,
+                        });
                     } else {
                         store.dispatch(P2pNetworkPubsubAction::OutgoingData {
-                            data: data.clone().into(),
+                            data: data.into(),
                             peer_id,
                         });
                     }
                 }
             }
+            P2pNetworkPubsubAction::OutgoingMessageError { .. } => {}
             P2pNetworkPubsubAction::OutgoingData { mut data, peer_id } => {
                 let Some(state) = store
                     .state()
@@ -217,6 +221,7 @@ impl P2pNetworkPubsubAction {
                     });
                 }
             }
+            P2pNetworkPubsubAction::SignError { .. } => (),
         }
     }
 }

p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs

@@ -106,7 +106,10 @@ impl P2pNetworkPubsubState {
                             }
                         }
                         for message in v.publish {
-                            let message_id = self.mcache.put(message.clone());
+                            let Some(message_id) = self.mcache.put(message.clone()) else {
+                                continue;
+                            };
+
                             let topic = self.topics.entry(message.topic.clone()).or_default();
                             if let Some(signature) = &message.signature {
                                 // skip recently seen message
@@ -258,6 +261,7 @@ impl P2pNetworkPubsubState {
                     v.message.control = None;
                 }
             }
+            P2pNetworkPubsubAction::OutgoingMessageError { .. } => {}
             P2pNetworkPubsubAction::Broadcast { .. } => {}
             P2pNetworkPubsubAction::Sign {
                 seqno,
@@ -277,6 +281,9 @@ impl P2pNetworkPubsubState {
                     key: None,
                 });
             }
+            P2pNetworkPubsubAction::SignError { .. } => {
+                let _ = self.to_sign.pop_front();
+            }
             P2pNetworkPubsubAction::BroadcastSigned { signature } => {
                 if let Some(mut message) = self.to_sign.pop_front() {
                     message.signature = Some(signature.clone().0.to_vec());

p2p/src/network/pubsub/p2p_network_pubsub_state.rs

@@ -43,38 +43,40 @@ pub struct P2pNetworkPubsubMessageCache {
 impl P2pNetworkPubsubMessageCache {
     const CAPACITY: usize = 100;
 
-    pub fn put(&mut self, message: pb::Message) -> Vec<u8> {
-        let id = compute_message_id(&message);
+    pub fn put(&mut self, message: pb::Message) -> Option<Vec<u8>> {
+        let id = compute_message_id(&message)?;
         self.map.insert(id.clone(), message);
         self.queue.push_back(id.clone());
         if self.queue.len() > Self::CAPACITY {
             if let Some(id) = self.queue.pop_back() {
                 self.map.remove(&id);
             }
         }
-        id
+        Some(id)
     }
 }
 
 // TODO: what if wasm32?
 // How to test it?
-pub fn compute_message_id(message: &pb::Message) -> Vec<u8> {
+pub fn compute_message_id(message: &pb::Message) -> Option<Vec<u8>> {
     let source_bytes = message
         .from
         .as_ref()
         .map(AsRef::as_ref)
         .unwrap_or(&[0, 1, 0][..]);
+
     let mut source_string = libp2p_identity::PeerId::from_bytes(source_bytes)
-        .expect("Valid peer id")
+        .ok()?
         .to_base58();
+
     let sequence_number = message
         .seqno
         .as_ref()
         .and_then(|b| <[u8; 8]>::try_from(b.as_slice()).ok())
         .map(u64::from_be_bytes)
         .unwrap_or_default();
     source_string.push_str(&sequence_number.to_string());
-    source_string.into_bytes()
+    Some(source_string.into_bytes())
 }
 
 #[derive(Default, Serialize, Deserialize, Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]