fix(p2p/libp2p/pubsub): duplicate block broadcasted from different senders for libp2p received blocks

Same issue affects transactions and snarks too, so needs to be fixed there as well.

Author: binier

node/src/transition_frontier/sync/transition_frontier_sync_effects.rs

@@ -322,11 +322,6 @@ impl TransitionFrontierSyncAction {
                 if !store.dispatch(TransitionFrontierSyncAction::BlocksNextApplyInit) {
                     store.dispatch(TransitionFrontierSyncAction::BlocksSuccess);
                 }
-
-                // TODO this should be handled by a callback
-                store.dispatch(P2pNetworkPubsubAction::BroadcastValidatedMessage {
-                    message_id: p2p::BroadcastMessageId::BlockHash { hash: hash.clone() },
-                });
             }
             TransitionFrontierSyncAction::BlocksSuccess => {}
             // Bootstrap/Catchup is practically complete at this point.

node/src/transition_frontier/transition_frontier_effects.rs

@@ -307,10 +307,18 @@ fn synced_effects<S: crate::Service>(
             best_tip: best_tip.block.clone(),
         });
     }
-    // rebroadcast block if received from webrtc network, otherwise noop.
-    store.dispatch(P2pNetworkPubsubAction::WebRtcRebroadcast {
-        message: GossipNetMessageV2::NewState(best_tip.block().clone()),
-    });
+    // TODO this should be handled by a callback
+    // If this get dispatched, we received block from libp2p.
+    if !store.dispatch(P2pNetworkPubsubAction::BroadcastValidatedMessage {
+        message_id: p2p::BroadcastMessageId::BlockHash {
+            hash: best_tip.hash().clone(),
+        },
+    }) {
+        // Otherwise block was received from WebRTC so inject it in libp2p.
+        store.dispatch(P2pNetworkPubsubAction::WebRtcRebroadcast {
+            message: GossipNetMessageV2::NewState(best_tip.block().clone()),
+        });
+    }
 
     let best_tip_hash = best_tip.merkle_root_hash().clone();
     store.dispatch(ConsensusAction::Prune);