Added broadcast delay for snarks and transactions until they are fully vertified

Author: 0xMimir

node/src/snark_pool/candidate/snark_pool_candidate_actions.rs

@@ -49,6 +49,7 @@ pub enum SnarkPoolCandidateAction {
     WorkVerifyError {
         peer_id: PeerId,
         verify_id: SnarkWorkVerifyId,
+        batch: Vec<Snark>,
     },
     WorkVerifySuccess {
         peer_id: PeerId,

node/src/snark_pool/candidate/snark_pool_candidate_reducer.rs

@@ -5,7 +5,7 @@ use openmina_core::snark::Snark;
 use p2p::{
     channels::rpc::{P2pChannelsRpcAction, P2pRpcId, P2pRpcRequest},
     disconnection::{P2pDisconnectionAction, P2pDisconnectionReason},
-    PeerId,
+    BroadcastMessageId, P2pNetworkPubsubAction, PeerId,
 };
 use snark::{work_verify::SnarkWorkVerifyAction, work_verify_effectful::SnarkWorkVerifyId};
 
@@ -124,10 +124,11 @@ impl SnarkPoolCandidatesState {
                             }
                         }),
                     on_error: redux::callback!(
-                        on_snark_pool_candidate_work_verify_error((req_id: SnarkWorkVerifyId, sender: String)) -> crate::Action {
+                        on_snark_pool_candidate_work_verify_error((req_id: SnarkWorkVerifyId, sender: String, batch: Vec<Snark>)) -> crate::Action {
                             SnarkPoolCandidateAction::WorkVerifyError {
                                 peer_id: sender.parse().unwrap(),
                                 verify_id: req_id,
+                                batch
                             }
                         }),
                 });
@@ -144,7 +145,11 @@ impl SnarkPoolCandidatesState {
             } => {
                 state.verify_pending(meta.time(), peer_id, *verify_id, job_ids);
             }
-            SnarkPoolCandidateAction::WorkVerifyError { peer_id, verify_id } => {
+            SnarkPoolCandidateAction::WorkVerifyError {
+                peer_id,
+                verify_id,
+                batch,
+            } => {
                 state.verify_result(meta.time(), peer_id, *verify_id, Err(()));
 
                 // TODO(binier): blacklist peer
@@ -154,6 +159,16 @@ impl SnarkPoolCandidatesState {
                     peer_id,
                     reason: P2pDisconnectionReason::SnarkPoolVerifyError,
                 });
+
+                for snark in batch {
+                    dispatcher.push(P2pNetworkPubsubAction::RejectMessage {
+                        message_id: Some(BroadcastMessageId::Snark {
+                            job_id: snark.job_id(),
+                        }),
+                        peer_id: None,
+                        reason: "Snark work verification failed".to_string(),
+                    });
+                }
             }
             SnarkPoolCandidateAction::WorkVerifySuccess {
                 peer_id,

node/src/snark_pool/snark_pool_reducer.rs

@@ -1,8 +1,11 @@
 use std::collections::BTreeMap;
 
 use openmina_core::snark::{SnarkJobCommitment, SnarkJobId};
-use p2p::channels::{
-    snark::P2pChannelsSnarkAction, snark_job_commitment::P2pChannelsSnarkJobCommitmentAction,
+use p2p::{
+    channels::{
+        snark::P2pChannelsSnarkAction, snark_job_commitment::P2pChannelsSnarkJobCommitmentAction,
+    },
+    BroadcastMessageId, P2pNetworkPubsubAction,
 };
 
 use crate::{snark_pool::JobCommitment, ExternalSnarkWorkerAction, SnarkerStrategy};
@@ -202,13 +205,17 @@ impl SnarkPoolState {
                     }
                 }
 
-                // TODO: libp2p logic already broadcasts everything right now and doesn't
-                // wait for validation, thad needs to be fixed. See #952
                 dispatcher.push(P2pChannelsSnarkAction::Libp2pBroadcast {
                     snark: snark.clone(),
                     nonce: 0,
                     is_local: *is_sender_local,
                 });
+
+                dispatcher.push(P2pNetworkPubsubAction::BroadcastValidatedMessage {
+                    message_id: BroadcastMessageId::Snark {
+                        job_id: snark.job_id(),
+                    },
+                });
             }
             SnarkPoolAction::P2pSendAll { .. } => {
                 let (dispatcher, global_state) = state_context.into_dispatcher_and_state();

node/src/transaction_pool/transaction_pool_actions.rs

@@ -8,7 +8,10 @@ use ledger::{
     },
     Account, AccountId,
 };
-use mina_p2p_messages::{list::List, v2};
+use mina_p2p_messages::{
+    list::List,
+    v2::{self, TransactionHash},
+};
 use openmina_core::{requests::RpcId, transaction::TransactionWithHash, ActionEvent};
 use redux::Callback;
 use serde::{Deserialize, Serialize};
@@ -38,6 +41,7 @@ pub enum TransactionPoolAction {
     #[action_event(level = warn, fields(debug(errors)))]
     VerifyError {
         errors: Vec<String>,
+        tx_hashes: Vec<TransactionHash>,
     },
     BestTipChanged {
         best_tip_hash: v2::LedgerHash,

node/src/transaction_pool/transaction_pool_reducer.rs

@@ -6,13 +6,16 @@ use ledger::{
     },
     Account, AccountId,
 };
+use mina_p2p_messages::v2::TransactionHash;
 use openmina_core::{
     bug_condition,
     constants::constraint_constants,
     requests::RpcId,
     transaction::{Transaction, TransactionWithHash},
 };
-use p2p::channels::transaction::P2pChannelsTransactionAction;
+use p2p::{
+    channels::transaction::P2pChannelsTransactionAction, BroadcastMessageId, P2pNetworkPubsubAction,
+};
 use redux::callback;
 use snark::user_command_verify::{SnarkUserCommandVerifyAction, SnarkUserCommandVerifyId};
 use std::collections::{BTreeMap, BTreeSet};
@@ -97,6 +100,12 @@ impl TransactionPoolState {
                     panic!()
                 };
 
+                let tx_hashes = commands
+                    .iter()
+                    .map(TransactionWithHash::hash)
+                    .cloned()
+                    .collect::<Vec<_>>();
+
                 // TODO: Convert those commands only once
                 let Ok(commands) = commands
                     .iter()
@@ -133,10 +142,11 @@ impl TransactionPoolState {
                             ),
                             on_error: callback!(
                                 on_snark_user_command_verify_error(
-                                    (req_id: SnarkUserCommandVerifyId, errors: Vec<String>)
+                                    (req_id: SnarkUserCommandVerifyId, errors: Vec<String>, tx_hashes: Vec<TransactionHash>)
                                 ) -> crate::Action {
                                     TransactionPoolAction::VerifyError {
-                                        errors
+                                        errors,
+                                        tx_hashes
                                     }
                                 }
                             )
@@ -147,6 +157,7 @@ impl TransactionPoolState {
                             let dispatcher = state.into_dispatcher();
                             dispatcher.push(TransactionPoolAction::VerifyError {
                                 errors: errors.clone(),
+                                tx_hashes,
                             });
                             if let Some(rpc_id) = from_rpc {
                                 dispatcher.push(RpcAction::TransactionInjectFailure {
@@ -186,8 +197,16 @@ impl TransactionPoolState {
                     from_rpc: *from_rpc,
                 });
             }
-            TransactionPoolAction::VerifyError { .. } => {
+            TransactionPoolAction::VerifyError { tx_hashes, .. } => {
                 // just logging the errors
+                let dispatcher = state.into_dispatcher();
+                for tx in tx_hashes {
+                    dispatcher.push(P2pNetworkPubsubAction::RejectMessage {
+                        message_id: Some(BroadcastMessageId::Transaction { tx: tx.clone() }),
+                        peer_id: None,
+                        reason: "Transaction rejected".to_owned(),
+                    });
+                }
             }
             TransactionPoolAction::BestTipChanged { best_tip_hash } => {
                 let account_ids = substate.pool.get_accounts_to_revalidate_on_new_best_tip();
@@ -300,9 +319,22 @@ impl TransactionPoolState {
                 if let Some(rpc_action) = rpc_action {
                     dispatcher.push(rpc_action);
                 }
-                // TODO: libp2p logic already broadcasts everything right now and doesn't
-                // wait for validation, thad needs to be fixed. See #952
+
                 if was_accepted {
+                    for tx in &accepted {
+                        dispatcher.push(P2pNetworkPubsubAction::BroadcastValidatedMessage {
+                            message_id: BroadcastMessageId::Transaction {
+                                tx: tx.hash.clone(),
+                            },
+                        });
+                    }
+                    for (tx, _) in &rejected {
+                        dispatcher.push(P2pNetworkPubsubAction::BroadcastValidatedMessage {
+                            message_id: BroadcastMessageId::Transaction {
+                                tx: tx.hash.clone(),
+                            },
+                        });
+                    }
                     dispatcher.push(TransactionPoolAction::Rebroadcast {
                         accepted,
                         rejected,

p2p/src/network/pubsub/mod.rs

@@ -18,6 +18,8 @@ mod p2p_network_pubsub_reducer;
 const TOPIC: &str = "coda/consensus-messages/0.0.1";
 
 pub mod pubsub_effectful;
+use mina_p2p_messages::v2::TransactionHash;
+use openmina_core::snark::SnarkJobId;
 pub use pubsub_effectful::P2pNetworkPubsubEffectfulAction;
 
 use binprot::BinProtWrite;
@@ -32,6 +34,12 @@ pub enum BroadcastMessageId {
     BlockHash {
         hash: mina_p2p_messages::v2::StateHash,
     },
+    Snark {
+        job_id: SnarkJobId,
+    },
+    Transaction {
+        tx: TransactionHash,
+    },
     MessageId {
         message_id: P2pNetworkPubsubMessageCacheId,
     },

p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs

@@ -1,11 +1,16 @@
-use std::{collections::btree_map::Entry, time::Duration};
+use std::{
+    collections::{btree_map::Entry, BTreeMap},
+    time::Duration,
+};
 
 use binprot::BinProtRead;
 use mina_p2p_messages::{
     gossip::{self, GossipNetMessageV2},
     v2::NetworkPoolSnarkPoolDiffVersionedStableV2,
 };
-use openmina_core::{block::BlockWithHash, bug_condition, fuzz_maybe, fuzzed_maybe, Substate};
+use openmina_core::{
+    block::BlockWithHash, bug_condition, fuzz_maybe, fuzzed_maybe, snark::Snark, Substate,
+};
 use redux::{Dispatcher, Timestamp};
 
 use crate::{
@@ -21,8 +26,8 @@ use super::{
         P2pNetworkPubsubMessageCacheMessage,
     },
     pb::{self, Message},
-    P2pNetworkPubsubAction, P2pNetworkPubsubClientState, P2pNetworkPubsubEffectfulAction,
-    P2pNetworkPubsubMessageCacheId, P2pNetworkPubsubState, TOPIC,
+    BroadcastMessageId, P2pNetworkPubsubAction, P2pNetworkPubsubClientState,
+    P2pNetworkPubsubEffectfulAction, P2pNetworkPubsubMessageCacheId, P2pNetworkPubsubState, TOPIC,
 };
 
 const MAX_MESSAGE_KEEP_DURATION: Duration = Duration::from_secs(300);
@@ -555,6 +560,39 @@ impl P2pNetworkPubsubState {
                             time,
                         }
                     }
+                    GossipNetMessageV2::SnarkPoolDiff {
+                        message: NetworkPoolSnarkPoolDiffVersionedStableV2::AddSolvedWork(snark),
+                        ..
+                    } => {
+                        let snark: Snark = snark.1.clone().into();
+                        let job_id = snark.job_id();
+                        P2pNetworkPubsubMessageCacheMessage::PreValidatedSnark {
+                            job_id,
+                            message,
+                            peer_id,
+                            time,
+                        }
+                    }
+                    GossipNetMessageV2::TransactionPoolDiff {
+                        message: tx_message,
+                        ..
+                    } => {
+                        let mut tx_hashes = BTreeMap::new();
+
+                        for tx in &tx_message.0 {
+                            let Ok(hash) = tx.hash() else {
+                                continue;
+                            };
+                            tx_hashes.insert(hash, false);
+                        }
+
+                        P2pNetworkPubsubMessageCacheMessage::PreValidatedTransactions {
+                            tx_hashes,
+                            message,
+                            peer_id,
+                            time,
+                        }
+                    }
                     _ => P2pNetworkPubsubMessageCacheMessage::PreValidated {
                         message,
                         peer_id,
@@ -572,7 +610,6 @@ impl P2pNetworkPubsubState {
                     GossipNetMessageV2::NewState(block) => {
                         let best_tip = BlockWithHash::try_new(block.clone())?;
                         dispatcher.push(P2pPeerAction::BestTipUpdate { peer_id, best_tip });
-                        return Ok(());
                     }
                     GossipNetMessageV2::TransactionPoolDiff { message, nonce } => {
                         let nonce = nonce.as_u32();
@@ -588,18 +625,15 @@ impl P2pNetworkPubsubState {
                         message: NetworkPoolSnarkPoolDiffVersionedStableV2::AddSolvedWork(work),
                         nonce,
                     } => {
+                        let snark: Snark = work.1.into();
                         dispatcher.push(P2pChannelsSnarkAction::Libp2pReceived {
                             peer_id,
-                            snark: Box::new(work.1.into()),
+                            snark: Box::new(snark),
                             nonce: nonce.as_u32(),
                         });
                     }
                     _ => {}
                 }
-
-                dispatcher.push(P2pNetworkPubsubAction::BroadcastValidatedMessage {
-                    message_id: super::BroadcastMessageId::MessageId { message_id },
-                });
                 Ok(())
             }
             P2pNetworkPubsubAction::BroadcastValidatedMessage { message_id } => {
@@ -609,8 +643,32 @@ impl P2pNetworkPubsubState {
                     bug_condition!("Message with id: {:?} not found", message_id);
                     return Ok(());
                 };
+
+                if let BroadcastMessageId::Transaction { tx } = &message_id {
+                    if let P2pNetworkPubsubMessageCacheMessage::PreValidatedTransactions {
+                        tx_hashes,
+                        ..
+                    } = message
+                    {
+                        if let Some(value) = tx_hashes.get_mut(tx) {
+                            *value = true;
+                        } else {
+                            bug_condition!("Transaction with hash: {} not found", tx);
+                            return Ok(());
+                        }
+
+                        let all_vertified = tx_hashes.values().all(|v| *v);
+                        if !all_vertified {
+                            return Ok(());
+                        }
+                    } else {
+                        bug_condition!("Invalid state for message id with type transaction");
+                        return Ok(());
+                    }
+                }
+
                 let raw_message = message.message().clone();
-                let peer_id = message.peer_id();
+                let peer_id = *message.peer_id();
 
                 pubsub_state.reduce_incoming_validated_message(
                     mcache_message_id,
@@ -628,7 +686,7 @@ impl P2pNetworkPubsubState {
                 *message = P2pNetworkPubsubMessageCacheMessage::Validated {
                     message: raw_message,
                     peer_id,
-                    time: message.time(),
+                    time: *message.time(),
                 };
 
                 let (dispatcher, state) = state_context.into_dispatcher_and_state();
@@ -641,7 +699,7 @@ impl P2pNetworkPubsubState {
                     .map
                     .iter()
                     .filter_map(|(message_id, message)| {
-                        if message.time() + MAX_MESSAGE_KEEP_DURATION > time {
+                        if (*message.time() + MAX_MESSAGE_KEEP_DURATION) <= time {
                             Some(message_id.to_owned())
                         } else {
                             None
@@ -669,7 +727,7 @@ impl P2pNetworkPubsubState {
                     };
 
                     if peer_id.is_none() {
-                        peer_id = Some(message.peer_id());
+                        peer_id = Some(*message.peer_id());
                     }
 
                     pubsub_state.mcache.remove_message(_message_id);