RPC Server: pushTransaction and pushHighPriorityTransaction metho…

…ds broadcast the transaction instead of only pushing it into the local mempool

By only pushing the transaction into the local mempool, the transaction never gets adopted if the node itself isn't an elected block producer.

lib/src/extras/rpc_server.rs

@@ -49,7 +49,7 @@ pub fn initialize_rpc_server(
     ));
     dispatcher.add(NetworkDispatcher::new(client.network()));
     if let Some(mempool) = client.mempool() {
-        dispatcher.add(MempoolDispatcher::new(mempool));
+        dispatcher.add(MempoolDispatcher::new(client.consensus_proxy(), mempool));
     }
     dispatcher.add(PolicyDispatcher {});
     if let Some(validator_proxy) = client.validator_proxy() {

mempool/src/mempool.rs

@@ -612,7 +612,7 @@ impl Mempool {
         (txs, size)
     }
 
-    /// Adds a transaction to the Mempool.
+    /// Adds a transaction to the local mempool without broadcasting it over the network.
     pub fn add_transaction(
         &self,
         transaction: Transaction,

rpc-interface/src/mempool.rs

@@ -9,11 +9,11 @@ use crate::types::{HashOrTx, MempoolInfo, RPCResult};
 pub trait MempoolInterface {
     type Error;
 
-    /// Pushes a raw transaction into the mempool, it will be assigned a default priority.
+    /// Pushes a raw transaction with a default priority assigned into the mempool and broadcast it to the network.
     async fn push_transaction(&mut self, raw_tx: String)
         -> RPCResult<Blake2bHash, (), Self::Error>;
 
-    /// Pushes a raw transaction into the mempool with high priority.
+    /// Pushes a raw transaction with a high priority assigned into the mempool and broadcast it to the network.
     async fn push_high_priority_transaction(
         &mut self,
         raw_tx: String,

rpc-server/src/dispatchers/mempool.rs

@@ -1,8 +1,10 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use nimiq_consensus::ConsensusProxy;
 use nimiq_hash::{Blake2bHash, Hash};
 use nimiq_mempool::{mempool::Mempool, mempool_transactions::TxPriority};
+use nimiq_network_libp2p::Network;
 use nimiq_rpc_interface::{
     mempool::MempoolInterface,
     types::{HashOrTx, MempoolInfo, RPCResult},
@@ -14,12 +16,13 @@ use crate::error::Error;
 
 #[allow(dead_code)]
 pub struct MempoolDispatcher {
+    consensus: ConsensusProxy<Network>,
     mempool: Arc<Mempool>,
 }
 
 impl MempoolDispatcher {
-    pub fn new(mempool: Arc<Mempool>) -> Self {
-        MempoolDispatcher { mempool }
+    pub fn new(consensus: ConsensusProxy<Network>, mempool: Arc<Mempool>) -> Self {
+        MempoolDispatcher { consensus, mempool }
     }
 }
 
@@ -35,8 +38,13 @@ impl MempoolInterface for MempoolDispatcher {
         let tx = Transaction::deserialize_from_vec(&hex::decode(&raw_tx)?)?;
         let txid = tx.hash::<Blake2bHash>();
 
-        match self.mempool.add_transaction(tx, None) {
-            Ok(_) => Ok(txid.into()),
+        match self.mempool.add_transaction(tx.clone(), None) {
+            Ok(_) => self
+                .consensus
+                .send_transaction(tx)
+                .await
+                .map(|_| txid.into())
+                .map_err(Error::NetworkError),
             Err(e) => Err(Error::MempoolError(e)),
         }
     }
@@ -48,8 +56,16 @@ impl MempoolInterface for MempoolDispatcher {
         let tx = Transaction::deserialize_from_vec(&hex::decode(&raw_tx)?)?;
         let txid = tx.hash::<Blake2bHash>();
 
-        match self.mempool.add_transaction(tx, Some(TxPriority::High)) {
-            Ok(_) => Ok(txid.into()),
+        match self
+            .mempool
+            .add_transaction(tx.clone(), Some(TxPriority::High))
+        {
+            Ok(_) => self
+                .consensus
+                .send_transaction(tx)
+                .await
+                .map(|_| txid.into())
+                .map_err(Error::NetworkError),
             Err(e) => Err(Error::MempoolError(e)),
         }
     }