[r2r] broadcast tx to txhlp / refactor tx errors

mm2src/coins/eth.rs

@@ -1119,6 +1119,17 @@ impl MarketCoinOps for EthCoin {
         )
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        let bytes = try_fus!(hex::decode(tx));
+        Box::new(
+            self.web3
+                .eth()
+                .send_raw_transaction(bytes.into())
+                .map(|res| format!("{:02x}", res))
+                .map_err(|e| ERRL!("{}", e)),
+        )
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/lightning.rs

@@ -398,6 +398,15 @@ impl MarketCoinOps for LightningCoin {
         ))
     }
 
+    fn send_raw_tx_bytes(&self, _tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        Box::new(futures01::future::err(
+            MmError::new(
+                "send_raw_tx is not supported for lightning, please use send_payment method instead.".to_string(),
+            )
+            .to_string(),
+        ))
+    }
+
     // Todo: Implement this when implementing swaps for lightning as it's is used mainly for swaps
     fn wait_for_confirmations(
         &self,

mm2src/coins/lp_coins.rs

@@ -422,6 +422,8 @@ pub trait MarketCoinOps {
     /// Receives raw transaction bytes in hexadecimal format as input and returns tx hash in hexadecimal format
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send>;
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send>;
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/qrc20.rs

@@ -1057,6 +1057,10 @@ impl MarketCoinOps for Qrc20Coin {
         utxo_common::send_raw_tx(&self.utxo, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/solana.rs

@@ -373,6 +373,20 @@ impl MarketCoinOps for SolanaCoin {
         Box::new(fut.boxed().compat())
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        let coin = self.clone();
+        let tx = tx.to_owned();
+        let fut = async move {
+            let bytes = hex::decode(tx).map_to_mm(|e| e).map_err(|e| format!("{:?}", e))?;
+            let tx: Transaction = deserialize(bytes.as_slice())
+                .map_to_mm(|e| e)
+                .map_err(|e| format!("{:?}", e))?;
+            let signature = coin.rpc().send_transaction(&tx).await.map_err(|e| format!("{:?}", e))?;
+            Ok(signature.to_string())
+        };
+        Box::new(fut.boxed().compat())
+    }
+
     fn wait_for_confirmations(
         &self,
         _tx: &[u8],

mm2src/coins/solana/solana_tests.rs

@@ -257,9 +257,15 @@ mod tests {
             .await
             .unwrap();
         println!("{:?}", valid_tx_details);
+
         let tx_str = str::from_utf8(&*valid_tx_details.tx_hex.0).unwrap();
         let res = sol_coin.send_raw_tx(tx_str).compat().await;
         assert_eq!(res.is_err(), false);
+
+        let res2 = sol_coin.send_raw_tx_bytes(&*valid_tx_details.tx_hex).compat().await;
+        assert_eq!(res2.is_err(), false);
+        assert_eq!(res, res2);
+
         //println!("{:?}", res);
     }
 

mm2src/coins/solana/spl.rs

@@ -222,6 +222,10 @@ impl MarketCoinOps for SplToken {
         self.platform_coin.send_raw_tx(tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        self.platform_coin.send_raw_tx_bytes(tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         _tx: &[u8],

mm2src/coins/solana/spl_tests.rs

@@ -82,9 +82,17 @@ mod tests {
         assert_eq!(valid_tx_details.my_balance_change, withdraw_amount.neg());
         assert_eq!(valid_tx_details.coin, "USDC".to_string());
         assert_ne!(valid_tx_details.timestamp, 0);
+
         let tx_str = from_utf8(&*valid_tx_details.tx_hex.0).unwrap();
         let res = usdc_sol_coin.send_raw_tx(tx_str).compat().await;
         assert_eq!(res.is_err(), false);
         println!("{:?}", res);
+
+        let res2 = usdc_sol_coin
+            .send_raw_tx_bytes(&*valid_tx_details.tx_hex.0)
+            .compat()
+            .await;
+        assert_eq!(res2.is_err(), false);
+        assert_eq!(res, res2);
     }
 }

mm2src/coins/test_coin.rs

@@ -44,6 +44,8 @@ impl MarketCoinOps for TestCoin {
     /// Receives raw transaction bytes in hexadecimal format as input and returns tx hash in hexadecimal format
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> { unimplemented!() }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> { unimplemented!() }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/bch.rs

@@ -1057,6 +1057,10 @@ impl MarketCoinOps for BchCoin {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/qtum.rs

@@ -677,6 +677,10 @@ impl MarketCoinOps for QtumCoin {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/slp.rs

@@ -1083,17 +1083,11 @@ impl MarketCoinOps for SlpToken {
 
     /// Receives raw transaction bytes in hexadecimal format as input and returns tx hash in hexadecimal format
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
-        let coin = self.clone();
-        let tx = tx.to_owned();
-        let fut = async move {
-            let bytes = hex::decode(tx).map_to_mm(|e| e).map_err(|e| format!("{:?}", e))?;
-            let tx = deserialize(bytes.as_slice())
-                .map_to_mm(|e| e)
-                .map_err(|e| format!("{:?}", e))?;
-            let hash = coin.broadcast_tx(&tx).await.map_err(|e| format!("{:?}", e))?;
-            Ok(format!("{:?}", hash))
-        };
-        Box::new(fut.boxed().compat())
+        utxo_common::send_raw_tx(self.as_ref(), tx)
+    }
+
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(self.as_ref(), tx)
     }
 
     fn wait_for_confirmations(
@@ -2022,7 +2016,7 @@ mod slp_tests {
 
         let err = fusd.send_raw_tx(&tx_bytes_str).wait().unwrap_err();
         println!("{:?}", err);
-        assert!(err.contains("is not valid with reason outputs greater than inputs"));
+        assert!(err.contains("the transaction was rejected by network rules.\\n\\ntransaction already in block chain"));
 
         let utxo_tx: UtxoTx = deserialize(tx_bytes).unwrap();
         let err = block_on(fusd.broadcast_tx(&utxo_tx)).unwrap_err();

mm2src/coins/utxo/utxo_common.rs

@@ -1696,7 +1696,10 @@ where
     Box::new(fut.boxed().compat())
 }
 
-pub fn send_raw_tx(coin: &UtxoCoinFields, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
+pub fn send_raw_tx<T: AsRef<[u8]>>(
+    coin: &UtxoCoinFields,
+    tx: T,
+) -> Box<dyn Future<Item = String, Error = String> + Send> {
     let bytes = try_fus!(hex::decode(tx));
     Box::new(
         coin.rpc_client

mm2src/coins/utxo/utxo_standard.rs

@@ -453,6 +453,10 @@ impl MarketCoinOps for UtxoStandardCoin {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/z_coin.rs

@@ -733,6 +733,10 @@ impl MarketCoinOps for ZCoin {
         utxo_common::send_raw_tx(self.as_ref(), tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(self.as_ref(), tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/lp_network.rs

@@ -16,11 +16,12 @@
 //  lp_network.rs
 //  marketmaker
 //
+use coins::lp_coinfind;
 use common::executor::spawn;
-use common::log;
 use common::mm_ctx::{MmArc, MmWeak};
 use common::mm_error::prelude::*;
 use common::mm_metrics::{ClockOps, MetricsOps};
+use common::{log, Future01CompatExt};
 use derive_more::Display;
 use futures::{channel::oneshot, StreamExt};
 use mm2_libp2p::atomicdex_behaviour::{AdexBehaviourCmd, AdexBehaviourEvent, AdexCmdTx, AdexEventRx, AdexResponse,
@@ -137,6 +138,16 @@ async fn process_p2p_message(
                 lp_swap::process_msg(ctx.clone(), split.next().unwrap_or_default(), &message.data).await;
                 to_propagate = true;
             },
+            Some(lp_swap::TX_HELPER_PREFIX) => {
+                if let Some(pair) = split.next() {
+                    if let Ok(Some(coin)) = lp_coinfind(&ctx, pair).await {
+                        match coin.send_raw_tx_bytes(&message.data).compat().await {
+                            Ok(id) => log::info!("Transaction broadcasted successfully: {:?} ", id),
+                            Err(e) => log::error!("Broadcast transaction failed. {}", e),
+                        }
+                    }
+                }
+            },
             None | Some(_) => (),
         }
     }
@@ -149,6 +160,7 @@ async fn process_p2p_message(
             &message.data,
             i_am_relay,
         );
+
         if process_fut.await {
             to_propagate = true;
         }

mm2src/lp_swap.rs

@@ -114,6 +114,8 @@ pub use trade_preimage::trade_preimage_rpc;
 
 pub const SWAP_PREFIX: TopicPrefix = "swap";
 
+pub const TX_HELPER_PREFIX: TopicPrefix = "txhlp";
+
 cfg_wasm32! {
     use common::indexed_db::{ConstructibleDb, DbLocked};
     use swap_wasm_db::{InitDbResult, SwapDb};
@@ -188,6 +190,16 @@ pub fn broadcast_swap_message(ctx: &MmArc, topic: String, msg: SwapMsg, p2p_priv
     broadcast_p2p_msg(ctx, vec![topic], encoded_msg);
 }
 
+pub fn broadcast_transaction_message(ctx: &MmArc, topic: String, msg: Vec<u8>, p2p_privkey: &Option<H256Json>) {
+    let p2p_private = match p2p_privkey {
+        Some(privkey) => privkey.0,
+        None => ctx.secp256k1_key_pair.or(&&|| panic!()).private().secret.take(),
+    };
+
+    let encoded_msg = encode_and_sign(&msg, &p2p_private).unwrap();
+    broadcast_p2p_msg(ctx, vec![topic], encoded_msg);
+}
+
 pub async fn process_msg(ctx: MmArc, topic: &str, msg: &[u8]) {
     let uuid = match Uuid::from_str(topic) {
         Ok(u) => u,
@@ -235,6 +247,16 @@ pub async fn process_msg(ctx: MmArc, topic: &str, msg: &[u8]) {
 
 pub fn swap_topic(uuid: &Uuid) -> String { pub_sub_topic(SWAP_PREFIX, &uuid.to_string()) }
 
+/// Formats and returns a topic format for `txhlp`.
+///
+/// # Usage
+/// ```ignore
+/// let topic = tx_helper_topic("BTC");
+/// // Returns topic format `txhlp/BTC` as String type.
+/// ```
+#[inline(always)]
+pub fn tx_helper_topic(coin: &str) -> String { pub_sub_topic(TX_HELPER_PREFIX, coin) }
+
 async fn recv_swap_msg<T>(
     ctx: MmArc,
     mut getter: impl FnMut(&mut SwapMsgStore) -> Option<T>,

mm2src/lp_swap/maker_swap.rs

@@ -11,6 +11,7 @@ use super::{broadcast_my_swap_status, broadcast_swap_message_every, check_other_
 use crate::mm2::lp_dispatcher::{DispatcherContext, LpEvents};
 use crate::mm2::lp_network::subscribe_to_topic;
 use crate::mm2::lp_ordermatch::{MakerOrderBuilder, OrderConfirmationsSettings};
+use crate::mm2::lp_swap::{broadcast_transaction_message, tx_helper_topic};
 use crate::mm2::MM_VERSION;
 use bigdecimal::BigDecimal;
 use bitcrypto::dhash160;
@@ -680,6 +681,13 @@ impl MakerSwap {
             },
         };
 
+        broadcast_transaction_message(
+            &self.ctx,
+            tx_helper_topic(self.maker_coin.ticker()),
+            transaction.tx_hex(),
+            &self.p2p_privkey,
+        );
+
         let tx_hash = transaction.tx_hash();
         log!({ "Maker payment tx {:02x}", tx_hash });
 
@@ -853,6 +861,13 @@ impl MakerSwap {
             },
         };
 
+        broadcast_transaction_message(
+            &self.ctx,
+            tx_helper_topic(self.taker_coin.ticker()),
+            transaction.tx_hex(),
+            &self.p2p_privkey,
+        );
+
         let tx_hash = transaction.tx_hash();
         log!({ "Taker payment spend tx {:02x}", tx_hash });
         let tx_ident = TransactionIdentifier {
@@ -925,6 +940,14 @@ impl MakerSwap {
                 ]))
             },
         };
+
+        broadcast_transaction_message(
+            &self.ctx,
+            tx_helper_topic(self.maker_coin.ticker()),
+            transaction.tx_hex(),
+            &self.p2p_privkey,
+        );
+
         let tx_hash = transaction.tx_hash();
         log!({ "Maker payment refund tx {:02x}", tx_hash });
         let tx_ident = TransactionIdentifier {
@@ -1143,6 +1166,14 @@ impl MakerSwap {
             Ok(Some(FoundSwapTxSpend::Spent(_))) => {
                 log!("Warning: MakerPayment spent, but TakerPayment is not yet. Trying to spend TakerPayment");
                 let transaction = try_s!(try_spend_taker_payment(self, &secret_hash.0).await);
+
+                broadcast_transaction_message(
+                    &self.ctx,
+                    tx_helper_topic(self.taker_coin.ticker()),
+                    transaction.tx_hex(),
+                    &self.p2p_privkey,
+                );
+
                 Ok(RecoveredSwap {
                     action: RecoveredSwapAction::SpentOtherPayment,
                     coin: self.taker_coin.ticker().to_string(),
@@ -1178,6 +1209,13 @@ impl MakerSwap {
                         .await
                 );
 
+                broadcast_transaction_message(
+                    &self.ctx,
+                    tx_helper_topic(self.maker_coin.ticker()),
+                    transaction.tx_hex(),
+                    &self.p2p_privkey,
+                );
+
                 Ok(RecoveredSwap {
                     action: RecoveredSwapAction::RefundedMyPayment,
                     coin: self.maker_coin.ticker().to_string(),