feat(rpc): Add invalidateblock and reconsiderblock RPC methods

zebra-rpc/src/methods.rs

@@ -341,6 +341,24 @@ pub trait Rpc {
         address_strings: AddressStrings,
     ) -> Result<Vec<GetAddressUtxos>>;
 
+    /// Invalidates a block if it is not yet finalized, removing it from the non-finalized
+    /// state if it is present and rejecting it during contextual validation if it is submitted.
+    ///
+    /// # Parameters
+    ///
+    /// - `block_hash`: (hex-encoded block hash, required) The block hash to invalidate.
+    // TODO: Invalidate block hashes even if they're not present in the non-finalized state.
+    #[method(name = "invalidateblock")]
+    async fn invalidate_block(&self, block_hash: GetBlockHash) -> Result<()>;
+
+    /// Reconsiders a previously invalidated block if it exists in the cache of previously invalidated blocks.
+    ///
+    /// # Parameters
+    ///
+    /// - `block_hash`: (hex-encoded block hash, required) The block hash to reconsider.
+    #[method(name = "reconsiderblock")]
+    async fn reconsider_block(&self, block_hash: GetBlockHash) -> Result<()>;
+
     /// Stop the running zebrad process.
     ///
     /// # Notes
@@ -357,7 +375,7 @@ pub trait Rpc {
 
 /// RPC method implementations.
 #[derive(Clone)]
-pub struct RpcImpl<Mempool, State, Tip, AddressBook>
+pub struct RpcImpl<Mempool, State, ReadState, Tip, AddressBook>
 where
     Mempool: Service<
             mempool::Request,
@@ -369,14 +387,23 @@ where
         + 'static,
     Mempool::Future: Send,
     State: Service<
+            zebra_state::Request,
+            Response = zebra_state::Response,
+            Error = zebra_state::BoxError,
+        > + Clone
+        + Send
+        + Sync
+        + 'static,
+    State::Future: Send,
+    ReadState: Service<
             zebra_state::ReadRequest,
             Response = zebra_state::ReadResponse,
             Error = zebra_state::BoxError,
         > + Clone
         + Send
         + Sync
         + 'static,
-    State::Future: Send,
+    ReadState::Future: Send,
     Tip: ChainTip + Clone + Send + Sync + 'static,
     AddressBook: AddressBookPeers + Clone + Send + Sync + 'static,
 {
@@ -407,6 +434,9 @@ where
     /// A handle to the state service.
     state: State,
 
+    /// A handle to the state service.
+    read_state: ReadState,
+
     /// Allows efficient access to the best tip of the blockchain.
     latest_chain_tip: Tip,
 
@@ -425,7 +455,8 @@ where
 /// A type alias for the last event logged by the server.
 pub type LoggedLastEvent = watch::Receiver<Option<(String, tracing::Level, chrono::DateTime<Utc>)>>;
 
-impl<Mempool, State, Tip, AddressBook> Debug for RpcImpl<Mempool, State, Tip, AddressBook>
+impl<Mempool, State, ReadState, Tip, AddressBook> Debug
+    for RpcImpl<Mempool, State, ReadState, Tip, AddressBook>
 where
     Mempool: Service<
             mempool::Request,
@@ -437,14 +468,23 @@ where
         + 'static,
     Mempool::Future: Send,
     State: Service<
+            zebra_state::Request,
+            Response = zebra_state::Response,
+            Error = zebra_state::BoxError,
+        > + Clone
+        + Send
+        + Sync
+        + 'static,
+    State::Future: Send,
+    ReadState: Service<
             zebra_state::ReadRequest,
             Response = zebra_state::ReadResponse,
             Error = zebra_state::BoxError,
         > + Clone
         + Send
         + Sync
         + 'static,
-    State::Future: Send,
+    ReadState::Future: Send,
     Tip: ChainTip + Clone + Send + Sync + 'static,
     AddressBook: AddressBookPeers + Clone + Send + Sync + 'static,
 {
@@ -460,7 +500,8 @@ where
     }
 }
 
-impl<Mempool, State, Tip, AddressBook> RpcImpl<Mempool, State, Tip, AddressBook>
+impl<Mempool, State, ReadState, Tip, AddressBook>
+    RpcImpl<Mempool, State, ReadState, Tip, AddressBook>
 where
     Mempool: Service<
             mempool::Request,
@@ -472,14 +513,23 @@ where
         + 'static,
     Mempool::Future: Send,
     State: Service<
+            zebra_state::Request,
+            Response = zebra_state::Response,
+            Error = zebra_state::BoxError,
+        > + Clone
+        + Send
+        + Sync
+        + 'static,
+    State::Future: Send,
+    ReadState: Service<
             zebra_state::ReadRequest,
             Response = zebra_state::ReadResponse,
             Error = zebra_state::BoxError,
         > + Clone
         + Send
         + Sync
         + 'static,
-    State::Future: Send,
+    ReadState::Future: Send,
     Tip: ChainTip + Clone + Send + Sync + 'static,
     AddressBook: AddressBookPeers + Clone + Send + Sync + 'static,
 {
@@ -496,6 +546,7 @@ where
         debug_like_zcashd: bool,
         mempool: Mempool,
         state: State,
+        read_state: ReadState,
         latest_chain_tip: Tip,
         address_book: AddressBook,
         last_warn_error_log_rx: LoggedLastEvent,
@@ -521,7 +572,8 @@ where
             debug_force_finished_sync,
             debug_like_zcashd,
             mempool: mempool.clone(),
-            state: state.clone(),
+            state,
+            read_state: read_state.clone(),
             latest_chain_tip: latest_chain_tip.clone(),
             queue_sender,
             address_book,
@@ -531,7 +583,7 @@ where
         // run the process queue
         let rpc_tx_queue_task_handle = tokio::spawn(
             runner
-                .run(mempool, state, latest_chain_tip, network)
+                .run(mempool, read_state, latest_chain_tip, network)
                 .in_current_span(),
         );
 
@@ -540,7 +592,8 @@ where
 }
 
 #[async_trait]
-impl<Mempool, State, Tip, AddressBook> RpcServer for RpcImpl<Mempool, State, Tip, AddressBook>
+impl<Mempool, State, ReadState, Tip, AddressBook> RpcServer
+    for RpcImpl<Mempool, State, ReadState, Tip, AddressBook>
 where
     Mempool: Service<
             mempool::Request,
@@ -552,14 +605,23 @@ where
         + 'static,
     Mempool::Future: Send,
     State: Service<
+            zebra_state::Request,
+            Response = zebra_state::Response,
+            Error = zebra_state::BoxError,
+        > + Clone
+        + Send
+        + Sync
+        + 'static,
+    State::Future: Send,
+    ReadState: Service<
             zebra_state::ReadRequest,
             Response = zebra_state::ReadResponse,
             Error = zebra_state::BoxError,
         > + Clone
         + Send
         + Sync
         + 'static,
-    State::Future: Send,
+    ReadState::Future: Send,
     Tip: ChainTip + Clone + Send + Sync + 'static,
     AddressBook: AddressBookPeers + Clone + Send + Sync + 'static,
 {
@@ -590,7 +652,8 @@ where
 
         let relay_fee = zebra_chain::transaction::zip317::MIN_MEMPOOL_TX_FEE_RATE as f64
             / (zebra_chain::amount::COIN as f64);
-        let difficulty = chain_tip_difficulty(self.network.clone(), self.state.clone()).await?;
+        let difficulty =
+            chain_tip_difficulty(self.network.clone(), self.read_state.clone()).await?;
 
         let response = GetInfo {
             version,
@@ -618,11 +681,11 @@ where
 
         let (usage_info_rsp, tip_pool_values_rsp, chain_tip_difficulty) = {
             use zebra_state::ReadRequest::*;
-            let state_call = |request| self.state.clone().oneshot(request);
+            let state_call = |request| self.read_state.clone().oneshot(request);
             tokio::join!(
                 state_call(UsageInfo),
                 state_call(TipPoolValues),
-                chain_tip_difficulty(network.clone(), self.state.clone())
+                chain_tip_difficulty(network.clone(), self.read_state.clone())
             )
         };
 
@@ -739,7 +802,7 @@ where
     }
 
     async fn get_address_balance(&self, address_strings: AddressStrings) -> Result<AddressBalance> {
-        let state = self.state.clone();
+        let state = self.read_state.clone();
 
         let valid_addresses = address_strings.valid_addresses()?;
 
@@ -822,7 +885,7 @@ where
     // - use `height_from_signed_int()` to handle negative heights
     //   (this might be better in the state request, because it needs the state height)
     async fn get_block(&self, hash_or_height: String, verbosity: Option<u8>) -> Result<GetBlock> {
-        let mut state = self.state.clone();
+        let mut state = self.read_state.clone();
         let verbosity = verbosity.unwrap_or(1);
         let network = self.network.clone();
         let original_hash_or_height = hash_or_height.clone();
@@ -1002,7 +1065,7 @@ where
         hash_or_height: String,
         verbose: Option<bool>,
     ) -> Result<GetBlockHeader> {
-        let state = self.state.clone();
+        let state = self.read_state.clone();
         let verbose = verbose.unwrap_or(true);
         let network = self.network.clone();
 
@@ -1231,7 +1294,7 @@ where
         txid: String,
         verbose: Option<u8>,
     ) -> Result<GetRawTransaction> {
-        let mut state = self.state.clone();
+        let mut state = self.read_state.clone();
         let mut mempool = self.mempool.clone();
         let verbose = verbose.unwrap_or(0) != 0;
 
@@ -1298,7 +1361,7 @@ where
     // - use `height_from_signed_int()` to handle negative heights
     //   (this might be better in the state request, because it needs the state height)
     async fn z_get_treestate(&self, hash_or_height: String) -> Result<GetTreestate> {
-        let mut state = self.state.clone();
+        let mut state = self.read_state.clone();
         let network = self.network.clone();
 
         // Reference for the legacy error code:
@@ -1387,7 +1450,7 @@ where
         start_index: NoteCommitmentSubtreeIndex,
         limit: Option<NoteCommitmentSubtreeIndex>,
     ) -> Result<GetSubtrees> {
-        let mut state = self.state.clone();
+        let mut state = self.read_state.clone();
 
         const POOL_LIST: &[&str] = &["sapling", "orchard"];
 
@@ -1453,7 +1516,7 @@ where
     }
 
     async fn get_address_tx_ids(&self, request: GetAddressTxIdsRequest) -> Result<Vec<String>> {
-        let mut state = self.state.clone();
+        let mut state = self.read_state.clone();
         let latest_chain_tip = self.latest_chain_tip.clone();
 
         let start = Height(request.start);
@@ -1510,7 +1573,7 @@ where
         &self,
         address_strings: AddressStrings,
     ) -> Result<Vec<GetAddressUtxos>> {
-        let mut state = self.state.clone();
+        let mut state = self.read_state.clone();
         let mut response_utxos = vec![];
 
         let valid_addresses = address_strings.valid_addresses()?;
@@ -1562,6 +1625,24 @@ where
         Ok(response_utxos)
     }
 
+    async fn invalidate_block(&self, GetBlockHash(block_hash): GetBlockHash) -> Result<()> {
+        self.state
+            .clone()
+            .oneshot(zebra_state::Request::InvalidateBlock(block_hash))
+            .await
+            .map(|rsp| assert_eq!(rsp, zebra_state::Response::Committed(block_hash)))
+            .map_misc_error()
+    }
+
+    async fn reconsider_block(&self, GetBlockHash(block_hash): GetBlockHash) -> Result<()> {
+        self.state
+            .clone()
+            .oneshot(zebra_state::Request::ReconsiderBlock(block_hash))
+            .await
+            .map(|rsp| assert_eq!(rsp, zebra_state::Response::Committed(block_hash)))
+            .map_misc_error()
+    }
+
     fn stop(&self) -> Result<String> {
         #[cfg(not(target_os = "windows"))]
         if self.network.is_regtest() {

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

@@ -960,6 +960,14 @@ fn mock_services<Tip>(
             zebra_node_services::mempool::Response,
             zebra_test::mock_service::PropTestAssertion,
         >,
+        tower::buffer::Buffer<
+            zebra_test::mock_service::MockService<
+                zebra_state::Request,
+                zebra_state::Response,
+                zebra_test::mock_service::PropTestAssertion,
+            >,
+            zebra_state::Request,
+        >,
         tower::buffer::Buffer<
             zebra_test::mock_service::MockService<
                 zebra_state::ReadRequest,
@@ -978,6 +986,7 @@ where
 {
     let mempool = MockService::build().for_prop_tests();
     let state = MockService::build().for_prop_tests();
+    let read_state = MockService::build().for_prop_tests();
 
     let (_tx, rx) = tokio::sync::watch::channel(None);
     let (rpc, mempool_tx_queue) = RpcImpl::new(
@@ -987,11 +996,12 @@ where
         false,
         true,
         mempool.clone(),
-        Buffer::new(state.clone(), 1),
+        Buffer::new(state, 1),
+        Buffer::new(read_state.clone(), 1),
         chain_tip,
         MockAddressBookPeers::new(vec![]),
         rx,
     );
 
-    (mempool, state, rpc, mempool_tx_queue)
+    (mempool, read_state, rpc, mempool_tx_queue)
 }