[CHRON-9324] Send RabbitMQ error message on transaction error (#37)

* Send RabbitMQ error message on transaction error

* Add transaction error type field

* Use operation id as a correlation id

Author: mexskican

rabbitmq/src/client.rs

@@ -4,11 +4,12 @@ use common::try_spawn;
 use enclose::enclose;
 use ethcore::client::{BlockChainClient, BlockId, ChainNotify, ChainRouteType, NewBlocks};
 use ethcore::miner;
-use failure::{Error, ResultExt};
-use futures::future::lazy;
+use ethereum_types::H256;
+use failure::{format_err, Error};
+use futures::future::{err, lazy};
 use handler::{Handler, Sender};
 use parity_runtime::Executor;
-use rabbitmq_adaptor::{ConsumerResult, RabbitConnection, RabbitExt};
+use rabbitmq_adaptor::{ConsumerResult, DeliveryExt, RabbitConnection, RabbitExt};
 use serde::Deserialize;
 use serde_json;
 use std::sync::Arc;
@@ -21,7 +22,10 @@ use DEFAULT_REPLY_QUEUE;
 use LOG_TARGET;
 use NEW_BLOCK_EXCHANGE_NAME;
 use NEW_BLOCK_ROUTING_KEY;
+use OPERATION_ID;
 use PUBLIC_TRANSACTION_QUEUE;
+use TX_ERROR_EXCHANGE_NAME;
+use TX_ERROR_ROUTING_KEY;
 
 #[derive(Debug, Default, Clone, PartialEq)]
 pub struct RabbitMqConfig {
@@ -38,6 +42,20 @@ pub struct PubSubClient<C> {
 #[derive(Deserialize)]
 struct TransactionMessage {
 	pub data: Bytes,
+	pub transaction_hash: H256,
+}
+
+#[derive(Serialize)]
+struct TransactionErrorMessage {
+	pub transaction_hash: H256,
+	pub error_message: String,
+	pub error_type: ErrorType,
+}
+
+#[derive(Serialize)]
+pub enum ErrorType {
+	LocalTransactionError,
+	TransactionRejected,
 }
 
 impl<C: 'static + miner::BlockChainClient + BlockChainClient> PubSubClient<C> {
@@ -58,21 +76,42 @@ impl<C: 'static + miner::BlockChainClient + BlockChainClient> PubSubClient<C> {
 					.clone()
 					.register_consumer(
 						PUBLIC_TRANSACTION_QUEUE.to_string(),
-						enclose!(() move |message| {
-							Box::new(std::str::from_utf8(&message.data)
-								.map_err(Error::from)
-								.and_then(|payload| {
-									serde_json::from_str(payload)
-										.context("Could not deserialize AMQP message payload")
-										.map_err(Error::from)
-								})
-								.and_then(|transaction_message: TransactionMessage| {
-									sender_handler.send_transaction(transaction_message.data)
-										.context(format!("Failed to send transaction"))
-										.map_err(Error::from)
-								})
+						enclose!((rabbit) move |message| {
+							let operation_id = message.get_header(OPERATION_ID);
+							if operation_id.is_none() {
+								return Box::new(err(format_err!("Missing protocol-id header")));
+							}
+							let operation_id = operation_id.unwrap();
+							let payload = std::str::from_utf8(&message.data);
+							if payload.is_err() {
+								return Box::new(err(format_err!("Could not parse AMQP message")));
+							}
+							let payload = payload.unwrap();
+							let transaction_message = serde_json::from_str(payload);
+							if transaction_message.is_err() {
+								return Box::new(err(format_err!("Could not deserialize AMQP message payload")));
+							}
+							let transaction_message: TransactionMessage = transaction_message.unwrap();
+							let transaction_hash = transaction_message.transaction_hash;
+							Box::new(sender_handler.send_transaction(transaction_message.data)
+								.map(|_| ())
 								.into_future()
-								.map(|_| ConsumerResult::ACK))
+								.or_else(enclose!((rabbit) move |error| {
+									let tx_error = TransactionErrorMessage {
+										transaction_hash,
+										error_message: format!("{}", error),
+										error_type: ErrorType::LocalTransactionError,
+									};
+									let serialized_message = serde_json::to_string(&tx_error).unwrap();
+									rabbit.clone().publish(
+										TX_ERROR_EXCHANGE_NAME.to_string(),
+										TX_ERROR_ROUTING_KEY.to_string(),
+										serialized_message.into(),
+										vec![(OPERATION_ID.to_string(), operation_id)],
+									).map(|_| ())
+								}))
+								.map(|_| ConsumerResult::ACK)
+							)
 						}),
 					)
 					.map_err(Error::from)

rabbitmq/src/handler.rs

@@ -1,12 +1,11 @@
-use failure::{Error, ResultExt};
 use std::sync::Arc;
 
 use ethcore::client::BlockChainClient;
 use ethcore::miner::{self, MinerService};
 use ethereum_types::H256;
-use hex;
 use rlp::Rlp;
 use common_types::transaction::{PendingTransaction, SignedTransaction};
+use common_types::transaction::Error as TransactionError;
 use types::Bytes;
 
 /// A Sender which uses references to a client and miner in order to send transactions
@@ -24,20 +23,19 @@ impl<C, M> Sender<C, M> {
 }
 
 pub trait Handler: Sync + Send {
-	fn send_transaction(&self, payload: Bytes) -> Result<H256, Error>;
+	fn send_transaction(&self, payload: Bytes) -> Result<H256, TransactionError>;
 }
 
 impl<C: miner::BlockChainClient + BlockChainClient, M: MinerService> Handler for Sender<C, M> {
-	fn send_transaction(&self, raw: Bytes) -> Result<H256, Error> {
+	fn send_transaction(&self, raw: Bytes) -> Result<H256, TransactionError> {
 		let signed_transaction = Rlp::new(&raw.into_vec())
 			.as_val()
-			.map_err(Error::from)
-			.and_then(|tx| SignedTransaction::new(tx).map_err(Error::from))?;
+			.map_err(TransactionError::from)
+			.and_then(|tx| SignedTransaction::new(tx).map_err(TransactionError::from))?;
 		let pending_transaction: PendingTransaction = signed_transaction.into();
 		let hash = pending_transaction.transaction.hash();
 		self.miner
 			.import_claimed_local_transaction(&*self.client, pending_transaction, false)
-			.map_err(Error::from)
 			.map(|_| hash)
 	}
 }

rabbitmq/src/lib.rs

@@ -36,4 +36,7 @@ const DEFAULT_REPLY_QUEUE: &'static str = "BlockchainInterface.default";
 const LOG_TARGET: &'static str = "rabbitmq";
 const NEW_BLOCK_EXCHANGE_NAME: &'static str = "BlockchainInterface.Output";
 const NEW_BLOCK_ROUTING_KEY: &'static str = "interface.in.new-block";
+const OPERATION_ID: &'static str = "operation-id";
 const PUBLIC_TRANSACTION_QUEUE: &'static str = "BlockchainInterface.public-tx";
+const TX_ERROR_EXCHANGE_NAME: &'static str = "BlockchainInterface.TransactionError";
+const TX_ERROR_ROUTING_KEY: &'static str = "interface.out.public-tx.error";