Implement `sendrawtransaction` RPC (#3685)

* Stub `sendrawtransaction` RPC method

Register the RPC method, and stub an implementation that currently just
panics. The method has a single `String` parameter with the hexadecimal
string of the raw transaction's bytes and returns a
`SentTransactionHash` wrapper type that's just a hexadecimal `String` of
the sent transaction's hash.

* Add mempool service instance to `RpcImpl`

Use a type parameter to represent the mempool service using the
interface defined by `zebra-node-services`.

* Update test vector to use a mock mempool service

Update the test to be compatible with the changes to `RpcImpl`. The mock
mempool service is expected to not be used during the test.

* Use a `tower::Buffer` for the mempool service

Make it simpler to send requests to the service in a concurrent manner.

* Return a `Future` from `send_raw_transaction`

Make the call asynchronous.

* Implement `sendrawtransaction` RPC

Deserialize the transaction and send it to be queued for verification
and subsequent inclusion in the mempool.

* Test if mempool receives sent raw transaction

Use a mock service as the mempool service and check that it receives a
sent raw transaction.

* Test using non-hexadecimal string parameter

The method should return an error.

* Test with bytes that fail deserialization

Check that the method returns an invalid parameters error if the input
can't be deserialized as a `Transaction`.

* Test if mempool errors are forwarded to caller

Mempool service errors should be sent back to the remote caller as
server errors.

* Test transactions rejected by the mempool service

Transactions that are rejected by the mempool service should result in
a server error being sent to the caller.

* Improve error message

Add the word "structurally" to make it clear that the issue is in the
transaction's deserialization.

Co-authored-by: Deirdre Connolly <durumcrustulum@gmail.com>

* Add note regarding missing `allowhighfees` param.

The parameter isn't supported yet because `lightwalletd` doesn't use it.

* Update the documentation to be consistent

Follow the convention adopted by the `get_info` RPC method.

* Remove mempool service usage line

It contained incomplete information that's not really necessary.

Co-authored-by: Alfredo Garcia <oxarbitrage@gmail.com>

* Fix formatting

`rustfmt` was not executed on the file for the previous commit because I
had edited it on GitHub.

Co-authored-by: Deirdre Connolly <durumcrustulum@gmail.com>
Co-authored-by: Alfredo Garcia <oxarbitrage@gmail.com>

Cargo.lock

@@ -5788,15 +5788,21 @@ name = "zebra-rpc"
 version = "1.0.0-beta.0"
 dependencies = [
  "futures",
+ "hex",
  "hyper",
  "jsonrpc-core",
  "jsonrpc-derive",
  "jsonrpc-http-server",
+ "proptest",
  "serde",
+ "thiserror",
  "tokio",
+ "tower",
  "tracing",
  "tracing-futures",
+ "zebra-chain",
  "zebra-network",
+ "zebra-node-services",
  "zebra-test",
 ]
 

zebra-rpc/Cargo.toml

@@ -8,6 +8,8 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+zebra-chain = { path = "../zebra-chain" }
+zebra-node-services = { path = "../zebra-node-services" }
 
 zebra-network = { path = "../zebra-network" }
 
@@ -21,12 +23,18 @@ jsonrpc-derive = "18.0.0"
 jsonrpc-http-server = "18.0.0"
 
 tokio = { version = "1.16.1", features = ["time", "rt-multi-thread", "macros", "tracing"] }
+tower = "0.4.12"
 
 tracing = "0.1"
 tracing-futures = "0.2"
 
+hex = "0.4.3"
 serde = { version = "1", features = ["serde_derive"] }
 
 [dev-dependencies]
+proptest = "0.10.1"
+thiserror = "1.0.30"
+tokio = { version = "1.16.1", features = ["full", "test-util"] }
 
+zebra-chain = { path = "../zebra-chain", features = ["proptest-impl"] }
 zebra-test = { path = "../zebra-test/" }

zebra-rpc/src/methods.rs

@@ -6,10 +6,15 @@
 //! Some parts of the `zcashd` RPC documentation are outdated.
 //! So this implementation follows the `lightwalletd` client implementation.
 
-use jsonrpc_core::{self, Result};
+use futures::FutureExt;
+use hex::FromHex;
+use jsonrpc_core::{self, BoxFuture, Error, ErrorCode, Result};
 use jsonrpc_derive::rpc;
+use tower::{buffer::Buffer, ServiceExt};
 
+use zebra_chain::{serialization::ZcashDeserialize, transaction::Transaction};
 use zebra_network::constants::USER_AGENT;
+use zebra_node_services::{mempool, BoxError};
 
 #[cfg(test)]
 mod tests;
@@ -48,15 +53,55 @@ pub trait Rpc {
     ///       note any other lightwalletd changes
     #[rpc(name = "getblockchaininfo")]
     fn get_blockchain_info(&self) -> Result<GetBlockChainInfo>;
+
+    /// sendrawtransaction
+    ///
+    /// Sends the raw bytes of a signed transaction to the network, if the transaction is valid.
+    ///
+    /// zcashd reference: <https://zcash.github.io/rpc/sendrawtransaction.html>
+    ///
+    /// Result: a hexadecimal string of the hash of the sent transaction.
+    ///
+    /// Note: zcashd provides an extra `allowhighfees` parameter, but we don't yet because
+    /// lightwalletd doesn't use it.
+    #[rpc(name = "sendrawtransaction")]
+    fn send_raw_transaction(
+        &self,
+        raw_transaction_hex: String,
+    ) -> BoxFuture<Result<SentTransactionHash>>;
 }
 
 /// RPC method implementations.
-
-pub struct RpcImpl {
+pub struct RpcImpl<Mempool>
+where
+    Mempool: tower::Service<mempool::Request, Response = mempool::Response, Error = BoxError>,
+{
     /// Zebra's application version.
-    pub app_version: String,
+    app_version: String,
+
+    /// A handle to the mempool service.
+    mempool: Buffer<Mempool, mempool::Request>,
 }
-impl Rpc for RpcImpl {
+
+impl<Mempool> RpcImpl<Mempool>
+where
+    Mempool: tower::Service<mempool::Request, Response = mempool::Response, Error = BoxError>,
+{
+    /// Create a new instance of the RPC handler.
+    pub fn new(app_version: String, mempool: Buffer<Mempool, mempool::Request>) -> Self {
+        RpcImpl {
+            app_version,
+            mempool,
+        }
+    }
+}
+
+impl<Mempool> Rpc for RpcImpl<Mempool>
+where
+    Mempool:
+        tower::Service<mempool::Request, Response = mempool::Response, Error = BoxError> + 'static,
+    Mempool::Future: Send,
+{
     fn get_info(&self) -> Result<GetInfo> {
         let response = GetInfo {
             build: self.app_version.clone(),
@@ -74,6 +119,53 @@ impl Rpc for RpcImpl {
 
         Ok(response)
     }
+
+    fn send_raw_transaction(
+        &self,
+        raw_transaction_hex: String,
+    ) -> BoxFuture<Result<SentTransactionHash>> {
+        let mempool = self.mempool.clone();
+
+        async move {
+            let raw_transaction_bytes = Vec::from_hex(raw_transaction_hex).map_err(|_| {
+                Error::invalid_params("raw transaction is not specified as a hex string")
+            })?;
+            let raw_transaction = Transaction::zcash_deserialize(&*raw_transaction_bytes)
+                .map_err(|_| Error::invalid_params("raw transaction is structurally invalid"))?;
+
+            let transaction_hash = raw_transaction.hash();
+
+            let transaction_parameter = mempool::Gossip::Tx(raw_transaction.into());
+            let request = mempool::Request::Queue(vec![transaction_parameter]);
+
+            let response = mempool.oneshot(request).await.map_err(|error| Error {
+                code: ErrorCode::ServerError(0),
+                message: error.to_string(),
+                data: None,
+            })?;
+
+            let queue_results = match response {
+                mempool::Response::Queued(results) => results,
+                _ => unreachable!("incorrect response variant from mempool service"),
+            };
+
+            assert_eq!(
+                queue_results.len(),
+                1,
+                "mempool service returned more results than expected"
+            );
+
+            match &queue_results[0] {
+                Ok(()) => Ok(SentTransactionHash(transaction_hash.to_string())),
+                Err(error) => Err(Error {
+                    code: ErrorCode::ServerError(0),
+                    message: error.to_string(),
+                    data: None,
+                }),
+            }
+        }
+        .boxed()
+    }
 }
 
 #[derive(serde::Serialize, serde::Deserialize)]
@@ -89,3 +181,9 @@ pub struct GetBlockChainInfo {
     chain: String,
     // TODO: add other fields used by lightwalletd (#3143)
 }
+
+#[derive(Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
+/// Response to a `sendrawtransaction` RPC request.
+///
+/// A JSON string with the transaction hash in hexadecimal.
+pub struct SentTransactionHash(String);

zebra-rpc/src/methods/tests.rs

@@ -1,2 +1,4 @@
 //! Test code for RPC methods
+
+mod prop;
 mod vectors;

zebra-rpc/src/methods/tests/prop.rs

@@ -0,0 +1,224 @@
+use jsonrpc_core::{Error, ErrorCode};
+use proptest::prelude::*;
+use thiserror::Error;
+use tower::buffer::Buffer;
+
+use zebra_chain::{
+    serialization::{ZcashDeserialize, ZcashSerialize},
+    transaction::{Transaction, UnminedTx},
+};
+use zebra_node_services::mempool;
+use zebra_test::mock_service::MockService;
+
+use super::super::{Rpc, RpcImpl, SentTransactionHash};
+
+proptest! {
+    /// Test that when sending a raw transaction, it is received by the mempool service.
+    #[test]
+    fn mempool_receives_raw_transaction(transaction in any::<Transaction>()) {
+        let runtime = zebra_test::init_async();
+
+        runtime.block_on(async move {
+            let mut mempool = MockService::build().for_prop_tests();
+            let rpc = RpcImpl::new("RPC test".to_owned(), Buffer::new(mempool.clone(), 1));
+            let hash = SentTransactionHash(transaction.hash().to_string());
+
+            let transaction_bytes = transaction
+                .zcash_serialize_to_vec()
+                .expect("Transaction serializes successfully");
+            let transaction_hex = hex::encode(&transaction_bytes);
+
+            let send_task = tokio::spawn(rpc.send_raw_transaction(transaction_hex));
+
+            let unmined_transaction = UnminedTx::from(transaction);
+            let expected_request = mempool::Request::Queue(vec![unmined_transaction.into()]);
+            let response = mempool::Response::Queued(vec![Ok(())]);
+
+            mempool
+                .expect_request(expected_request)
+                .await?
+                .respond(response);
+
+            let result = send_task
+                .await
+                .expect("Sending raw transactions should not panic");
+
+            prop_assert_eq!(result, Ok(hash));
+
+            Ok::<_, TestCaseError>(())
+        })?;
+    }
+
+    /// Test that mempool errors are forwarded to the caller.
+    ///
+    /// Mempool service errors should become server errors.
+    #[test]
+    fn mempool_errors_are_forwarded(transaction in any::<Transaction>()) {
+        let runtime = zebra_test::init_async();
+
+        runtime.block_on(async move {
+            let mut mempool = MockService::build().for_prop_tests();
+            let rpc = RpcImpl::new("RPC test".to_owned(), Buffer::new(mempool.clone(), 1));
+
+            let transaction_bytes = transaction
+                .zcash_serialize_to_vec()
+                .expect("Transaction serializes successfully");
+            let transaction_hex = hex::encode(&transaction_bytes);
+
+            let send_task = tokio::spawn(rpc.send_raw_transaction(transaction_hex));
+
+            let unmined_transaction = UnminedTx::from(transaction);
+            let expected_request = mempool::Request::Queue(vec![unmined_transaction.into()]);
+
+            mempool
+                .expect_request(expected_request)
+                .await?
+                .respond(Err(DummyError));
+
+            let result = send_task
+                .await
+                .expect("Sending raw transactions should not panic");
+
+            prop_assert!(
+                matches!(
+                    result,
+                    Err(Error {
+                        code: ErrorCode::ServerError(_),
+                        ..
+                    })
+                ),
+                "Result is not a server error: {result:?}"
+            );
+
+            Ok::<_, TestCaseError>(())
+        })?;
+    }
+
+    /// Test that when the mempool rejects a transaction the caller receives an error.
+    #[test]
+    fn rejected_transactions_are_reported(transaction in any::<Transaction>()) {
+        let runtime = zebra_test::init_async();
+
+        runtime.block_on(async move {
+            let mut mempool = MockService::build().for_prop_tests();
+            let rpc = RpcImpl::new("RPC test".to_owned(), Buffer::new(mempool.clone(), 1));
+
+            let transaction_bytes = transaction
+                .zcash_serialize_to_vec()
+                .expect("Transaction serializes successfully");
+            let transaction_hex = hex::encode(&transaction_bytes);
+
+            let send_task = tokio::spawn(rpc.send_raw_transaction(transaction_hex));
+
+            let unmined_transaction = UnminedTx::from(transaction);
+            let expected_request = mempool::Request::Queue(vec![unmined_transaction.into()]);
+            let response = mempool::Response::Queued(vec![Err(DummyError.into())]);
+
+            mempool
+                .expect_request(expected_request)
+                .await?
+                .respond(response);
+
+            let result = send_task
+                .await
+                .expect("Sending raw transactions should not panic");
+
+            prop_assert!(
+                matches!(
+                    result,
+                    Err(Error {
+                        code: ErrorCode::ServerError(_),
+                        ..
+                    })
+                ),
+                "Result is not a server error: {result:?}"
+            );
+
+            Ok::<_, TestCaseError>(())
+        })?;
+    }
+
+    /// Test that the method rejects non-hexadecimal characters.
+    ///
+    /// Try to call `send_raw_transaction` using a string parameter that has at least one
+    /// non-hexadecimal character, and check that it fails with an expected error.
+    #[test]
+    fn non_hexadecimal_string_results_in_an_error(non_hex_string in ".*[^0-9A-Fa-f].*") {
+        let runtime = zebra_test::init_async();
+        let _guard = runtime.enter();
+
+        // CORRECTNESS: Nothing in this test depends on real time, so we can speed it up.
+        tokio::time::pause();
+
+        runtime.block_on(async move {
+            let mut mempool = MockService::build().for_prop_tests();
+            let rpc = RpcImpl::new("RPC test".to_owned(), Buffer::new(mempool.clone(), 1));
+
+            let send_task = tokio::spawn(rpc.send_raw_transaction(non_hex_string));
+
+            mempool.expect_no_requests().await?;
+
+            let result = send_task
+                .await
+                .expect("Sending raw transactions should not panic");
+
+            prop_assert!(
+                matches!(
+                    result,
+                    Err(Error {
+                        code: ErrorCode::InvalidParams,
+                        ..
+                    })
+                ),
+                "Result is not an invalid parameters error: {result:?}"
+            );
+
+            Ok::<_, TestCaseError>(())
+        })?;
+    }
+
+    /// Test that the method rejects an input that's not a transaction.
+    ///
+    /// Try to call `send_raw_transaction` using random bytes that fail to deserialize as a
+    /// transaction, and check that it fails with an expected error.
+    #[test]
+    fn invalid_transaction_results_in_an_error(random_bytes in any::<Vec<u8>>()) {
+        let runtime = zebra_test::init_async();
+        let _guard = runtime.enter();
+
+        // CORRECTNESS: Nothing in this test depends on real time, so we can speed it up.
+        tokio::time::pause();
+
+        prop_assume!(Transaction::zcash_deserialize(&*random_bytes).is_err());
+
+        runtime.block_on(async move {
+            let mut mempool = MockService::build().for_prop_tests();
+            let rpc = RpcImpl::new("RPC test".to_owned(), Buffer::new(mempool.clone(), 1));
+
+            let send_task = tokio::spawn(rpc.send_raw_transaction(hex::encode(random_bytes)));
+
+            mempool.expect_no_requests().await?;
+
+            let result = send_task
+                .await
+                .expect("Sending raw transactions should not panic");
+
+            prop_assert!(
+                matches!(
+                    result,
+                    Err(Error {
+                        code: ErrorCode::InvalidParams,
+                        ..
+                    })
+                ),
+                "Result is not an invalid parameters error: {result:?}"
+            );
+
+            Ok::<_, TestCaseError>(())
+        })?;
+    }
+}
+
+#[derive(Clone, Copy, Debug, Error)]
+#[error("a dummy error type")]
+pub struct DummyError;

zebra-rpc/src/methods/tests/vectors.rs

@@ -1,15 +1,23 @@
 //! Fixed test vectors for RPC methods.
 
-use super::super::*;
-use zebra_network::constants::USER_AGENT;
+use tower::buffer::Buffer;
 
-#[test]
-fn rpc_getinfo() {
+use zebra_network::constants::USER_AGENT;
+use zebra_node_services::BoxError;
+use zebra_test::mock_service::MockService;
+
+use super::super::*;
+
+#[tokio::test]
+async fn rpc_getinfo() {
     zebra_test::init();
 
-    let rpc = RpcImpl {
-        app_version: "Zebra version test".to_string(),
-    };
+    let mut mempool: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests();
+
+    let rpc = RpcImpl::new(
+        "Zebra version test".to_string(),
+        Buffer::new(mempool.clone(), 1),
+    );
 
     let get_info = rpc.get_info().expect("We should have a GetInfo struct");
 
@@ -20,4 +28,6 @@ fn rpc_getinfo() {
     // make sure there is a `subversion` field,
     // and that is equal to the Zebra user agent.
     assert_eq!(get_info.subversion, USER_AGENT);
+
+    mempool.expect_no_requests().await;
 }

zebra-rpc/src/server.rs

@@ -4,11 +4,13 @@
 //! `"jsonrpc" = 1.0` fields in JSON-RPC 1.0 requests,
 //! such as `lightwalletd`.
 
+use jsonrpc_core;
+use jsonrpc_http_server::ServerBuilder;
+use tower::buffer::Buffer;
 use tracing::*;
 use tracing_futures::Instrument;
 
-use jsonrpc_core;
-use jsonrpc_http_server::ServerBuilder;
+use zebra_node_services::{mempool, BoxError};
 
 use crate::{
     config::Config,
@@ -24,12 +26,21 @@ pub struct RpcServer;
 
 impl RpcServer {
     /// Start a new RPC server endpoint
-    pub fn spawn(config: Config, app_version: String) -> tokio::task::JoinHandle<()> {
+    pub fn spawn<Mempool>(
+        config: Config,
+        app_version: String,
+        mempool: Buffer<Mempool, mempool::Request>,
+    ) -> tokio::task::JoinHandle<()>
+    where
+        Mempool: tower::Service<mempool::Request, Response = mempool::Response, Error = BoxError>
+            + 'static,
+        Mempool::Future: Send,
+    {
         if let Some(listen_addr) = config.listen_addr {
             info!("Trying to open RPC endpoint at {}...", listen_addr,);
 
             // Initialize the rpc methods with the zebra version
-            let rpc_impl = RpcImpl { app_version };
+            let rpc_impl = RpcImpl::new(app_version, mempool);
 
             // Create handler compatible with V1 and V2 RPC protocols
             let mut io =

zebrad/src/commands/start.rs

@@ -155,7 +155,8 @@ impl StartCmd {
             .service(mempool);
 
         // Launch RPC server
-        let rpc_task_handle = RpcServer::spawn(config.rpc, app_version().to_string());
+        let rpc_task_handle =
+            RpcServer::spawn(config.rpc, app_version().to_string(), mempool.clone());
 
         let setup_data = InboundSetupData {
             address_book,