darkfid/task/unknown_proposal: fixed erroenous subscription notification

bin/darkfid/src/proto/mod.rs

@@ -105,8 +105,9 @@ impl DarkfidP2pHandler {
         );
 
         // Start the `ProtocolProposal` messages handler
-        let subscriber = subscribers.get("proposals").unwrap().clone();
-        self.proposals.start(executor, validator, &self.p2p, subscriber).await?;
+        let proposals_sub = subscribers.get("proposals").unwrap().clone();
+        let blocks_sub = subscribers.get("blocks").unwrap().clone();
+        self.proposals.start(executor, validator, &self.p2p, proposals_sub, blocks_sub).await?;
 
         // Start the `ProtocolSync` messages handler
         self.sync.start(executor, validator).await?;

bin/darkfid/src/proto/protocol_proposal.rs

@@ -80,7 +80,8 @@ impl ProtocolProposalHandler {
         executor: &ExecutorPtr,
         validator: &ValidatorPtr,
         p2p: &P2pPtr,
-        subscriber: JsonSubscriber,
+        proposals_sub: JsonSubscriber,
+        blocks_sub: JsonSubscriber,
     ) -> Result<()> {
         debug!(
             target: "darkfid::proto::protocol_proposal::start",
@@ -88,7 +89,7 @@ impl ProtocolProposalHandler {
         );
 
         self.handler.task.clone().start(
-            handle_receive_proposal(self.handler.clone(), self.tasks.clone(), validator.clone(), p2p.clone(), subscriber, executor.clone()),
+            handle_receive_proposal(self.handler.clone(), self.tasks.clone(), validator.clone(), p2p.clone(), proposals_sub, blocks_sub, executor.clone()),
             |res| async move {
                 match res {
                     Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
@@ -127,7 +128,8 @@ async fn handle_receive_proposal(
     tasks: Arc<RwLock<HashSet<StoppableTaskPtr>>>,
     validator: ValidatorPtr,
     p2p: P2pPtr,
-    subscriber: JsonSubscriber,
+    proposals_sub: JsonSubscriber,
+    blocks_sub: JsonSubscriber,
     executor: ExecutorPtr,
 ) -> Result<()> {
     debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "START");
@@ -160,9 +162,9 @@ async fn handle_receive_proposal(
                 // Signal handler to broadcast the valid proposal to rest nodes
                 handler.send_action(channel, ProtocolGenericAction::Broadcast).await;
 
-                // Notify subscriber
+                // Notify proposals subscriber
                 let enc_prop = JsonValue::String(base64::encode(&serialize_async(&proposal).await));
-                subscriber.notify(vec![enc_prop].into()).await;
+                proposals_sub.notify(vec![enc_prop].into()).await;
 
                 continue
             }
@@ -186,7 +188,7 @@ async fn handle_receive_proposal(
         let _tasks = tasks.clone();
         let _task = task.clone();
         task.clone().start(
-            handle_unknown_proposal(validator.clone(), p2p.clone(), subscriber.clone(), channel, proposal.0),
+            handle_unknown_proposal(validator.clone(), p2p.clone(), proposals_sub.clone(), blocks_sub.clone(), channel, proposal.0),
             |res| async move {
                 match res {
                     Ok(()) | Err(Error::DetachedTaskStopped) => { _tasks.write().await.remove(&_task); }

bin/darkfid/src/task/unknown_proposal.rs

@@ -45,7 +45,8 @@ use crate::proto::{
 pub async fn handle_unknown_proposal(
     validator: ValidatorPtr,
     p2p: P2pPtr,
-    subscriber: JsonSubscriber,
+    proposals_sub: JsonSubscriber,
+    blocks_sub: JsonSubscriber,
     channel: u32,
     proposal: Proposal,
 ) -> Result<()> {
@@ -95,25 +96,25 @@ pub async fn handle_unknown_proposal(
     // Response should not be empty
     if response.proposals.is_empty() {
         warn!(target: "darkfid::task::handle_unknown_proposal", "Peer responded with empty sequence, node might be out of sync!");
-        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
+        return handle_reorg(validator, p2p, proposals_sub, blocks_sub, channel, proposal).await
     }
 
     // Sequence length must correspond to requested height
     if response.proposals.len() as u32 != proposal.block.header.height - last.0 {
         debug!(target: "darkfid::task::handle_unknown_proposal", "Response sequence length is erroneous");
-        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
+        return handle_reorg(validator, p2p, proposals_sub, blocks_sub, channel, proposal).await
     }
 
     // First proposal must extend canonical
     if response.proposals[0].block.header.previous != last.1 {
         debug!(target: "darkfid::task::handle_unknown_proposal", "Response sequence doesn't extend canonical");
-        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
+        return handle_reorg(validator, p2p, proposals_sub, blocks_sub, channel, proposal).await
     }
 
     // Last proposal must be the same as the one requested
     if response.proposals.last().unwrap().hash != proposal.hash {
         debug!(target: "darkfid::task::handle_unknown_proposal", "Response sequence doesn't correspond to requested tip");
-        return handle_reorg(validator, p2p, subscriber, channel, proposal).await
+        return handle_reorg(validator, p2p, proposals_sub, blocks_sub, channel, proposal).await
     }
 
     // Process response proposals
@@ -136,9 +137,9 @@ pub async fn handle_unknown_proposal(
         let message = ProposalMessage(proposal.clone());
         p2p.broadcast_with_exclude(&message, &[channel.address().clone()]).await;
 
-        // Notify subscriber
+        // Notify proposals subscriber
         let enc_prop = JsonValue::String(base64::encode(&serialize_async(proposal).await));
-        subscriber.notify(vec![enc_prop].into()).await;
+        proposals_sub.notify(vec![enc_prop].into()).await;
     }
 
     Ok(())
@@ -153,7 +154,8 @@ pub async fn handle_unknown_proposal(
 async fn handle_reorg(
     validator: ValidatorPtr,
     p2p: P2pPtr,
-    subscriber: JsonSubscriber,
+    proposals_sub: JsonSubscriber,
+    blocks_sub: JsonSubscriber,
     channel: ChannelPtr,
     proposal: Proposal,
 ) -> Result<()> {
@@ -480,19 +482,21 @@ async fn handle_reorg(
         }
     };
 
-    if confirmed.is_empty() {
-        return Ok(())
-    }
-
-    let mut notif_blocks = Vec::with_capacity(confirmed.len());
-    for block in confirmed {
-        notif_blocks.push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
+    if !confirmed.is_empty() {
+        let mut notif_blocks = Vec::with_capacity(confirmed.len());
+        for block in confirmed {
+            notif_blocks.push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
+        }
+        blocks_sub.notify(JsonValue::Array(notif_blocks)).await;
     }
-    subscriber.notify(JsonValue::Array(notif_blocks)).await;
 
     // Broadcast proposal to the network
-    let message = ProposalMessage(proposal);
+    let message = ProposalMessage(proposal.clone());
     p2p.broadcast(&message).await;
 
+    // Notify proposals subscriber
+    let enc_prop = JsonValue::String(base64::encode(&serialize_async(&proposal).await));
+    proposals_sub.notify(vec![enc_prop].into()).await;
+
     Ok(())
 }