Updated sources for transaction pool

Author: 0xMimir

Cargo.lock

@@ -36,6 +36,10 @@ mina-p2p-messages = { workspace = true }
 poseidon = { workspace = true }
 hex = "0.4.3"
 ark-ff = { workspace = true }
+libp2p-identity = { version = "=0.2.7", features = [
+    "serde",
+    "peerid"
+] }
 
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
 redux = { workspace = true, features = ["serializable_callbacks"] }

core/Cargo.toml

@@ -17,6 +17,7 @@ pub mod constants;
 pub mod dummy;
 
 pub mod block;
+pub mod p2p;
 pub mod snark;
 pub mod transaction;
 

core/src/lib.rs

@@ -0,0 +1,16 @@
+/// TODO: These types and methods should be moved to `p2p` crate, they are here because they are used in `snark` crates callbacks
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Copy)]
+pub struct P2pNetworkPubsubMessageCacheId {
+    pub source: libp2p_identity::PeerId,
+    pub seqno: u64,
+}
+
+impl P2pNetworkPubsubMessageCacheId {
+    pub fn to_raw_bytes(&self) -> Vec<u8> {
+        let mut message_id = self.source.to_base58();
+        message_id.push_str(&self.seqno.to_string());
+        message_id.into_bytes()
+    }
+}

core/src/p2p.rs

@@ -5,3 +5,36 @@ mod transaction_with_hash;
 pub use transaction_with_hash::*;
 
 pub use mina_p2p_messages::v2::{MinaBaseUserCommandStableV2 as Transaction, TransactionHash};
+
+use crate::{p2p::P2pNetworkPubsubMessageCacheId, requests::RpcId};
+
+/// TODO: Types and methods bellow, should be moved to `node` crate, they are here because they are used in `snark` crates callbacks
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, Copy, Default)]
+pub enum TransactionPoolMessageSource {
+    Rpc {
+        id: RpcId,
+    },
+    Pubsub {
+        id: P2pNetworkPubsubMessageCacheId,
+    },
+    #[default]
+    None,
+}
+
+impl TransactionPoolMessageSource {
+    pub fn rpc(id: RpcId) -> Self {
+        Self::Rpc { id }
+    }
+
+    pub fn pubsub(id: P2pNetworkPubsubMessageCacheId) -> Self {
+        Self::Pubsub { id }
+    }
+
+    pub fn is_sender_local(&self) -> bool {
+        matches!(self, Self::Rpc { .. })
+    }
+
+    pub fn is_libp2p(&self) -> bool {
+        matches!(self, Self::Pubsub { .. })
+    }
+}

core/src/transaction/mod.rs

@@ -2,7 +2,7 @@ use openmina_core::{
     block::AppliedBlock,
     bug_condition,
     requests::{RequestId, RpcId, RpcIdType},
-    transaction::TransactionWithHash,
+    transaction::{TransactionPoolMessageSource, TransactionWithHash},
 };
 use p2p::{
     connection::{incoming::P2pConnectionIncomingAction, outgoing::P2pConnectionOutgoingAction},
@@ -515,7 +515,7 @@ impl RpcState {
                 dispatcher.push(RpcAction::TransactionInjectPending { rpc_id: *rpc_id });
                 dispatcher.push(TransactionPoolAction::StartVerify {
                     commands: commands_with_hash,
-                    from_rpc: Some(*rpc_id),
+                    from_source: TransactionPoolMessageSource::rpc(*rpc_id),
                 });
             }
             RpcAction::TransactionInjectPending { rpc_id } => {

node/src/rpc/rpc_reducer.rs

@@ -160,6 +160,7 @@ impl SnarkPoolCandidatesState {
                     reason: P2pDisconnectionReason::SnarkPoolVerifyError,
                 });
 
+                // TODO: this is incorrect, only one snark in batch could be invalid but we would reject that whole batch
                 for snark_job_id in batch {
                     dispatcher.push(P2pNetworkPubsubAction::RejectMessage {
                         message_id: Some(BroadcastMessageId::Snark {

node/src/snark_pool/candidate/snark_pool_candidate_reducer.rs

@@ -1,15 +1,11 @@
 use std::collections::BTreeMap;
 
+use crate::{snark_pool::JobCommitment, ExternalSnarkWorkerAction, SnarkerStrategy};
 use openmina_core::snark::{SnarkJobCommitment, SnarkJobId};
-use p2p::{
-    channels::{
-        snark::P2pChannelsSnarkAction, snark_job_commitment::P2pChannelsSnarkJobCommitmentAction,
-    },
-    BroadcastMessageId, P2pNetworkPubsubAction,
+use p2p::channels::{
+    snark::P2pChannelsSnarkAction, snark_job_commitment::P2pChannelsSnarkJobCommitmentAction,
 };
 
-use crate::{snark_pool::JobCommitment, ExternalSnarkWorkerAction, SnarkerStrategy};
-
 use super::{
     JobState, SnarkPoolAction, SnarkPoolActionWithMetaRef, SnarkPoolEffectfulAction,
     SnarkPoolState, SnarkWork,
@@ -210,12 +206,6 @@ impl SnarkPoolState {
                     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/snark_pool/snark_pool_reducer.rs

@@ -4,7 +4,9 @@ use std::time::Duration;
 use malloc_size_of_derive::MallocSizeOf;
 use mina_p2p_messages::v2;
 use openmina_core::constants::PROTOCOL_VERSION;
-use openmina_core::transaction::{TransactionInfo, TransactionWithHash};
+use openmina_core::transaction::{
+    TransactionInfo, TransactionPoolMessageSource, TransactionWithHash,
+};
 use p2p::P2pNetworkPubsubMessageCacheId;
 use rand::prelude::*;
 
@@ -539,10 +541,10 @@ impl P2p {
                 }
             )),
             on_p2p_channels_transaction_libp2p_received: Some(redux::callback!(
-                on_p2p_channels_transaction_libp2p_received(transactions: Vec<TransactionWithHash>) -> crate::Action {
+                on_p2p_channels_transaction_libp2p_received((transactions: Vec<TransactionWithHash>, id: P2pNetworkPubsubMessageCacheId)) -> crate::Action {
                     TransactionPoolAction::StartVerify {
                         commands: transactions.into_iter().collect(),
-                        from_rpc: None
+                        from_source: TransactionPoolMessageSource::pubsub(id),
                     }
                 }
             )),

node/src/state.rs

@@ -1,6 +1,7 @@
 #![allow(clippy::unit_arg)]
 
 use crate::{p2p_ready, TransactionPoolAction};
+use openmina_core::transaction::TransactionPoolMessageSource;
 use p2p::{
     channels::rpc::{P2pChannelsRpcAction, P2pRpcId, P2pRpcRequest},
     PeerId,
@@ -103,7 +104,7 @@ impl TransactionPoolCandidatesState {
                 let transaction_hashes = batch.iter().map(|tx| tx.hash().clone()).collect();
                 dispatcher.push(TransactionPoolAction::StartVerify {
                     commands: batch.into_iter().collect(),
-                    from_rpc: None,
+                    from_source: TransactionPoolMessageSource::None,
                 });
                 dispatcher.push(TransactionPoolCandidateAction::VerifyPending {
                     peer_id,