feat(rpc): Implement an RPC transaction queue

zebra-rpc/src/lib.rs

@@ -6,6 +6,7 @@
 
 pub mod config;
 pub mod methods;
+pub mod queue;
 pub mod server;
 #[cfg(test)]
 mod tests;

zebra-rpc/src/methods.rs

@@ -21,11 +21,13 @@ use zebra_chain::{
     chain_tip::ChainTip,
     parameters::{ConsensusBranchId, Network, NetworkUpgrade},
     serialization::{SerializationError, ZcashDeserialize},
-    transaction::{self, SerializedTransaction, Transaction},
+    transaction::{self, SerializedTransaction, Transaction, UnminedTx},
 };
 use zebra_network::constants::USER_AGENT;
 use zebra_node_services::{mempool, BoxError};
 
+use crate::queue::{Queue, Runner};
+
 #[cfg(test)]
 mod tests;
 
@@ -160,17 +162,23 @@ where
     /// The configured network for this RPC service.
     #[allow(dead_code)]
     network: Network,
+
+    /// An instance of the RPC transaction queue
+    queue_runner: Runner,
 }
 
 impl<Mempool, State, Tip> RpcImpl<Mempool, State, Tip>
 where
-    Mempool: Service<mempool::Request, Response = mempool::Response, Error = BoxError>,
+    Mempool: Service<mempool::Request, Response = mempool::Response, Error = BoxError> + 'static,
     State: Service<
-        zebra_state::ReadRequest,
-        Response = zebra_state::ReadResponse,
-        Error = zebra_state::BoxError,
-    >,
-    Tip: ChainTip + Send + Sync,
+            zebra_state::ReadRequest,
+            Response = zebra_state::ReadResponse,
+            Error = zebra_state::BoxError,
+        > + Clone
+        + Send
+        + Sync
+        + 'static,
+    Tip: ChainTip + Clone + Send + Sync + 'static,
 {
     /// Create a new instance of the RPC handler.
     pub fn new<Version>(
@@ -182,14 +190,31 @@ where
     ) -> Self
     where
         Version: ToString,
+        <Mempool as Service<mempool::Request>>::Future: Send,
+        <State as Service<zebra_state::ReadRequest>>::Future: Send,
     {
-        RpcImpl {
+        let (mut listener, runner) = Queue::start();
+
+        let rpc_impl = RpcImpl {
             app_version: app_version.to_string(),
-            mempool,
-            state,
-            latest_chain_tip,
+            mempool: mempool.clone(),
+            state: state.clone(),
+            latest_chain_tip: latest_chain_tip.clone(),
             network,
-        }
+            queue_runner: runner.clone(),
+        };
+
+        // run the listener
+        tokio::spawn(async move {
+            listener.listen().await;
+        });
+
+        // run the process queue
+        tokio::spawn(async move {
+            runner.run(mempool, state, latest_chain_tip).await;
+        });
+
+        rpc_impl
     }
 }
 
@@ -319,6 +344,7 @@ where
         raw_transaction_hex: String,
     ) -> BoxFuture<Result<SentTransactionHash>> {
         let mempool = self.mempool.clone();
+        let queue_sender = self.queue_runner.sender();
 
         async move {
             let raw_transaction_bytes = Vec::from_hex(raw_transaction_hex).map_err(|_| {
@@ -329,6 +355,10 @@ where
 
             let transaction_hash = raw_transaction.hash();
 
+            // send transaction to the rpc queue
+            let unmined_transaction = UnminedTx::from(raw_transaction.clone());
+            let _ = queue_sender.send(Some(unmined_transaction)).await;
+
             let transaction_parameter = mempool::Gossip::Tx(raw_transaction.into());
             let request = mempool::Request::Queue(vec![transaction_parameter]);
 

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

@@ -512,6 +512,94 @@ proptest! {
             Ok::<_, TestCaseError>(())
         })?;
     }
+
+    /// Test the queue functionality using `send_raw_transaction`
+    #[test]
+    fn rpc_queue_main_loop(tx in any::<Transaction>())
+    {
+        let runtime = zebra_test::init_async();
+        let _guard = runtime.enter();
+
+        let transaction_hash = tx.hash();
+
+        runtime.block_on(async move {
+            tokio::time::pause();
+
+            let mut mempool = MockService::build().for_prop_tests();
+            let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
+
+            let rpc = RpcImpl::new(
+                "RPC test",
+                Buffer::new(mempool.clone(), 1),
+                Buffer::new(state.clone(), 1),
+                NoChainTip,
+                Mainnet,
+            );
+
+            // send a transaction
+            let tx_bytes = tx
+                .zcash_serialize_to_vec()
+                .expect("Transaction serializes successfully");
+            let tx_hex = hex::encode(&tx_bytes);
+            let send_task = tokio::spawn(rpc.send_raw_transaction(tx_hex));
+
+            let tx_unmined = UnminedTx::from(tx);
+            let expected_request = mempool::Request::Queue(vec![tx_unmined.clone().into()]);
+
+            // fail the mempool insertion
+            mempool
+                .expect_request(expected_request)
+                .await
+                .unwrap()
+                .respond(Err(DummyError));
+
+            let _ = send_task
+                .await
+                .expect("Sending raw transactions should not panic");
+
+            // make sure the transaction was inserted to the queue
+            prop_assert_eq!(rpc.queue_runner.queue().lock().unwrap().transactions().len(), 1);
+
+            // advance enough time to have a new runner iteration
+            let spacing = chrono::Duration::seconds(150);
+            tokio::time::advance(spacing.to_std().unwrap()).await;
+
+            // the runner will made a new call to TransactionsById
+            let mut transactions_hash_set = HashSet::new();
+            transactions_hash_set.insert(tx_unmined.id);
+            let expected_request = mempool::Request::TransactionsById(transactions_hash_set);
+            let response = mempool::Response::Transactions(vec![]);
+
+            mempool
+                .expect_request(expected_request)
+                .await?
+                .respond(response);
+
+            // the runner will also query the state again for the transaction
+            let expected_request = zebra_state::ReadRequest::Transaction(transaction_hash);
+            let response = zebra_state::ReadResponse::Transaction(None);
+
+            state
+                .expect_request(expected_request)
+                .await?
+                .respond(response);
+
+            // now a retry will be sent to the mempool
+            let expected_request = mempool::Request::Queue(vec![mempool::Gossip::Tx(tx_unmined.clone())]);
+            let response = mempool::Response::Queued(vec![Ok(())]);
+
+            mempool
+                .expect_request(expected_request)
+                .await?
+                .respond(response);
+
+            // no more requets are done
+            mempool.expect_no_requests().await?;
+            state.expect_no_requests().await?;
+
+            Ok::<_, TestCaseError>(())
+        })?;
+    }
 }
 
 #[derive(Clone, Copy, Debug, Error)]