fix(rpc): Check for panics in the transaction queue (#4046)

* Check for panics in the RPC transaction queue * Add missing pin! and abort in the start task * Check for transaction queue panics in tests
ZcashFoundation · Apr 6, 2022 · ecfb7ed · ecfb7ed
1 parent 4e3229b
commit ecfb7ed
Show file tree

Hide file tree

Showing 5 changed files with 126 additions and 41 deletions.
diff --git a/zebra-rpc/src/methods.rs b/zebra-rpc/src/methods.rs
@@ -14,8 +14,9 @@ use hex::{FromHex, ToHex};
 use indexmap::IndexMap;
 use jsonrpc_core::{self, BoxFuture, Error, ErrorCode, Result};
 use jsonrpc_derive::rpc;
-use tokio::sync::broadcast::Sender;
+use tokio::{sync::broadcast::Sender, task::JoinHandle};
 use tower::{buffer::Buffer, Service, ServiceExt};
+use tracing::Instrument;
 
 use zebra_chain::{
     block::{self, Height, SerializedBlock},
@@ -188,7 +189,7 @@ where
         state: State,
         latest_chain_tip: Tip,
         network: Network,
-    ) -> Self
+    ) -> (Self, JoinHandle<()>)
     where
         Version: ToString,
         <Mempool as Service<mempool::Request>>::Future: Send,
@@ -206,21 +207,13 @@ where
         };
 
         // run the process queue
-        let mut queue_task_handler = tokio::spawn(async move {
-            runner.run(mempool, state, latest_chain_tip, network).await;
-        });
-
-        // queue panic checker
-        tokio::spawn(async move {
-            loop {
-                let queue_task_handler = &mut queue_task_handler;
-                if queue_task_handler.await.is_err() {
-                    panic!("Unexpected panic in the RPC queue");
-                }
-            }
-        });
+        let rpc_tx_queue_task_handle = tokio::spawn(
+            runner
+                .run(mempool, state, latest_chain_tip, network)
+                .in_current_span(),
+        );
 
-        rpc_impl
+        (rpc_impl, rpc_tx_queue_task_handle)
     }
 }
 

diff --git a/zebra-rpc/src/methods/tests/prop.rs b/zebra-rpc/src/methods/tests/prop.rs
@@ -2,6 +2,7 @@
 
 use std::collections::HashSet;
 
+use futures::FutureExt;
 use hex::ToHex;
 use jsonrpc_core::{Error, ErrorCode};
 use proptest::prelude::*;
@@ -34,7 +35,7 @@ proptest! {
         runtime.block_on(async move {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -67,6 +68,10 @@ proptest! {
 
             prop_assert_eq!(result, Ok(hash));
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -82,7 +87,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -122,6 +127,10 @@ proptest! {
                 "Result is not a server error: {result:?}"
             );
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -135,7 +144,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -176,6 +185,10 @@ proptest! {
                 "Result is not a server error: {result:?}"
             );
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -196,7 +209,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -224,6 +237,10 @@ proptest! {
                 "Result is not an invalid parameters error: {result:?}"
             );
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -246,7 +263,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -274,6 +291,10 @@ proptest! {
                 "Result is not an invalid parameters error: {result:?}"
             );
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -295,7 +316,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -323,6 +344,10 @@ proptest! {
 
             prop_assert_eq!(result, Ok(expected_response));
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -343,7 +368,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -371,6 +396,10 @@ proptest! {
                 "Result is not an invalid parameters error: {result:?}"
             );
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -393,7 +422,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -420,6 +449,11 @@ proptest! {
                 ),
                 "Result is not an invalid parameters error: {result:?}"
             );
+
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -431,16 +465,23 @@ proptest! {
         let _guard = runtime.enter();
         let mut mempool = MockService::build().for_prop_tests();
         let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
+
         // look for an error with a `NoChainTip`
-        let rpc = RpcImpl::new(
+        let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
             "RPC test",
             Buffer::new(mempool.clone(), 1),
             Buffer::new(state.clone(), 1),
             NoChainTip,
             network,
         );
+
         let response = rpc.get_blockchain_info();
         prop_assert_eq!(&response.err().unwrap().message, "No Chain tip available yet");
+
+        // The queue task should continue without errors or panics
+        let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+        prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
         runtime.block_on(async move {
             mempool.expect_no_requests().await?;
             state.expect_no_requests().await?;
@@ -471,7 +512,7 @@ proptest! {
         mock_chain_tip_sender.send_best_tip_block_time(block_time);
 
         // Start RPC with the mocked `ChainTip`
-        let rpc = RpcImpl::new(
+        let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
             "RPC test",
             Buffer::new(mempool.clone(), 1),
             Buffer::new(state.clone(), 1),
@@ -504,6 +545,10 @@ proptest! {
             },
         };
 
+        // The queue task should continue without errors or panics
+        let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+        prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
         // check no requests were made during this test
         runtime.block_on(async move {
             mempool.expect_no_requests().await?;
@@ -528,7 +573,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -594,6 +639,10 @@ proptest! {
             mempool.expect_no_requests().await?;
             state.expect_no_requests().await?;
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }
@@ -611,7 +660,7 @@ proptest! {
             let mut mempool = MockService::build().for_prop_tests();
             let mut state: MockService<_, _, _, BoxError> = MockService::build().for_prop_tests();
 
-            let rpc = RpcImpl::new(
+            let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
                 "RPC test",
                 Buffer::new(mempool.clone(), 1),
                 Buffer::new(state.clone(), 1),
@@ -686,6 +735,10 @@ proptest! {
             mempool.expect_no_requests().await?;
             state.expect_no_requests().await?;
 
+            // The queue task should continue without errors or panics
+            let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+            prop_assert!(matches!(rpc_tx_queue_task_result, None));
+
             Ok::<_, TestCaseError>(())
         })?;
     }

diff --git a/zebra-rpc/src/methods/tests/vectors.rs b/zebra-rpc/src/methods/tests/vectors.rs
@@ -25,7 +25,7 @@ async fn rpc_getinfo() {
     let mut mempool: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests();
     let mut state: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests();
 
-    let rpc = RpcImpl::new(
+    let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
         "RPC test",
         Buffer::new(mempool.clone(), 1),
         Buffer::new(state.clone(), 1),
@@ -45,6 +45,10 @@ async fn rpc_getinfo() {
 
     mempool.expect_no_requests().await;
     state.expect_no_requests().await;
+
+    // The queue task should continue without errors or panics
+    let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+    assert!(matches!(rpc_tx_queue_task_result, None));
 }
 
 #[tokio::test]
@@ -63,7 +67,7 @@ async fn rpc_getblock() {
         zebra_state::populated_state(blocks.clone(), Mainnet).await;
 
     // Init RPC
-    let rpc = RpcImpl::new(
+    let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
         "RPC test",
         Buffer::new(mempool.clone(), 1),
         read_state,
@@ -82,6 +86,10 @@ async fn rpc_getblock() {
     }
 
     mempool.expect_no_requests().await;
+
+    // The queue task should continue without errors or panics
+    let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+    assert!(matches!(rpc_tx_queue_task_result, None));
 }
 
 #[tokio::test]
@@ -92,7 +100,7 @@ async fn rpc_getblock_error() {
     let mut state: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests();
 
     // Init RPC
-    let rpc = RpcImpl::new(
+    let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
         "RPC test",
         Buffer::new(mempool.clone(), 1),
         Buffer::new(state.clone(), 1),
@@ -108,6 +116,10 @@ async fn rpc_getblock_error() {
 
     mempool.expect_no_requests().await;
     state.expect_no_requests().await;
+
+    // The queue task should continue without errors or panics
+    let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+    assert!(matches!(rpc_tx_queue_task_result, None));
 }
 
 #[tokio::test]
@@ -132,7 +144,7 @@ async fn rpc_getbestblockhash() {
         zebra_state::populated_state(blocks.clone(), Mainnet).await;
 
     // Init RPC
-    let rpc = RpcImpl::new(
+    let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
         "RPC test",
         Buffer::new(mempool.clone(), 1),
         read_state,
@@ -150,6 +162,10 @@ async fn rpc_getbestblockhash() {
     assert_eq!(response_hash, tip_block_hash);
 
     mempool.expect_no_requests().await;
+
+    // The queue task should continue without errors or panics
+    let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+    assert!(matches!(rpc_tx_queue_task_result, None));
 }
 
 #[tokio::test]
@@ -168,7 +184,7 @@ async fn rpc_getrawtransaction() {
         zebra_state::populated_state(blocks.clone(), Mainnet).await;
 
     // Init RPC
-    let rpc = RpcImpl::new(
+    let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
         "RPC test",
         Buffer::new(mempool.clone(), 1),
         read_state,
@@ -231,4 +247,8 @@ async fn rpc_getrawtransaction() {
             }
         }
     }
+
+    // The queue task should continue without errors or panics
+    let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
+    assert!(matches!(rpc_tx_queue_task_result, None));
 }