change(state): Stop re-downloading blocks that are in non-finalized side chains (#6335)

* Adds 'Contains' request in state, and:

- adds finalized block hashes to sent_blocks

- replaces Depth call with Contains in sync, inbound, and block verifier

* removes unnecessary From impl

* Apply suggestions from code review

Co-authored-by: teor <teor@riseup.net>

* updates references to Request::Contains

* Renames zs::response::BlockLocation to KnownBlocks

* Updates AlreadyInChain error

* update docs for sent_hashes.add_finalized

* Update zebra-consensus/src/block.rs

Co-authored-by: teor <teor@riseup.net>

* Update comment for `sent_blocks` field in state service

* update KnownBlock request to check the non-finalized state before responding that a block is in the queue

* Apply suggestions from code review

Co-authored-by: teor <teor@riseup.net>

* update references to renamed method

* Clear sent_blocks when there's a reset

* Move self.finalized_block_write_sender.is_none() to can_fork_chain_at

* revert changes related to checking queue

---------

Co-authored-by: teor <teor@riseup.net>

zebra-consensus/src/block.rs

@@ -154,15 +154,15 @@ where
                 .ready()
                 .await
                 .map_err(|source| VerifyBlockError::Depth { source, hash })?
-                .call(zs::Request::Depth(hash))
+                .call(zs::Request::KnownBlock(hash))
                 .await
                 .map_err(|source| VerifyBlockError::Depth { source, hash })?
             {
-                zs::Response::Depth(Some(depth)) => {
+                zs::Response::KnownBlock(Some(location)) => {
-                    return Err(BlockError::AlreadyInChain(hash, depth).into())
+                    return Err(BlockError::AlreadyInChain(hash, location).into())
                 }
-                zs::Response::Depth(None) => {}
+                zs::Response::KnownBlock(None) => {}
-                _ => unreachable!("wrong response to Request::Depth"),
+                _ => unreachable!("wrong response to Request::KnownBlock"),
             }
 
             tracing::trace!("performing block checks");

zebra-consensus/src/error.rs

@@ -245,8 +245,8 @@ pub enum BlockError {
     #[error("block contains duplicate transactions")]
     DuplicateTransaction,
 
-    #[error("block {0:?} is already in the chain at depth {1:?}")]
+    #[error("block {0:?} is already in present in the state {1:?}")]
-    AlreadyInChain(zebra_chain::block::Hash, u32),
+    AlreadyInChain(zebra_chain::block::Hash, zebra_state::KnownBlock),
 
     #[error("invalid block {0:?}: missing block height")]
     MissingHeight(zebra_chain::block::Hash),

zebra-state/src/lib.rs

@@ -33,7 +33,7 @@ pub use config::{check_and_delete_old_databases, Config};
 pub use constants::MAX_BLOCK_REORG_HEIGHT;
 pub use error::{BoxError, CloneError, CommitBlockError, ValidateContextError};
 pub use request::{FinalizedBlock, HashOrHeight, PreparedBlock, ReadRequest, Request};
-pub use response::{ReadResponse, Response};
+pub use response::{KnownBlock, ReadResponse, Response};
 pub use service::{
     chain_tip::{ChainTipChange, LatestChainTip, TipAction},
     init, spawn_init,

zebra-state/src/request.rs

@@ -607,6 +607,13 @@ pub enum Request {
     /// * [`Response::BlockHash(None)`](Response::BlockHash) otherwise.
     BestChainBlockHash(block::Height),
 
+    /// Checks if a block is present anywhere in the state service.
+    /// Looks up `hash` in block queues as well as the finalized chain and all non-finalized chains.
+    ///
+    /// Returns [`Response::KnownBlock(Some(Location))`](Response::KnownBlock) if the block is in the best state service.
+    /// Returns [`Response::KnownBlock(None)`](Response::KnownBlock) otherwise.
+    KnownBlock(block::Hash),
+
     #[cfg(feature = "getblocktemplate-rpcs")]
     /// Performs contextual validation of the given block, but does not commit it to the state.
     ///
@@ -634,6 +641,7 @@ impl Request {
             }
             Request::BestChainNextMedianTimePast => "best_chain_next_median_time_past",
             Request::BestChainBlockHash(_) => "best_chain_block_hash",
+            Request::KnownBlock(_) => "known_block",
             #[cfg(feature = "getblocktemplate-rpcs")]
             Request::CheckBlockProposalValidity(_) => "check_block_proposal_validity",
         }
@@ -947,6 +955,8 @@ impl TryFrom<Request> for ReadRequest {
                      Manually convert the request to ReadRequest::AnyChainUtxo, \
                      and handle pending UTXOs"),
 
+            Request::KnownBlock(_) => Err("ReadService does not track queued blocks"),
+
             #[cfg(feature = "getblocktemplate-rpcs")]
             Request::CheckBlockProposalValidity(prepared) => {
                 Ok(ReadRequest::CheckBlockProposalValidity(prepared))

zebra-state/src/response.rs

@@ -69,11 +69,27 @@ pub enum Response {
     /// specified block hash.
     BlockHash(Option<block::Hash>),
 
+    /// Response to [`Request::KnownBlock`].
+    KnownBlock(Option<KnownBlock>),
+
     #[cfg(feature = "getblocktemplate-rpcs")]
     /// Response to [`Request::CheckBlockProposalValidity`](Request::CheckBlockProposalValidity)
     ValidBlockProposal,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+/// An enum of block stores in the state where a block hash could be found.
+pub enum KnownBlock {
+    /// Block is in the best chain.
+    BestChain,
+
+    /// Block is in a side chain.
+    SideChain,
+
+    /// Block is queued to be validated and committed, or rejected and dropped.
+    Queue,
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 /// A response to a read-only
 /// [`ReadStateService`](crate::service::ReadStateService)'s

zebra-state/src/service.rs

@@ -160,7 +160,7 @@ pub(crate) struct StateService {
     // - remove block hashes once their heights are strictly less than the finalized tip
     last_sent_finalized_block_hash: block::Hash,
 
-    /// A set of non-finalized block hashes that have been sent to the block write task.
+    /// A set of block hashes that have been sent to the block write task.
     /// Hashes of blocks below the finalized tip height are periodically pruned.
     sent_non_finalized_block_hashes: SentHashes,
 
@@ -713,13 +713,13 @@ impl StateService {
             return rsp_rx;
         }
 
-        // Wait until block commit task is ready to write non-finalized blocks before dequeuing them
         if self.finalized_block_write_sender.is_none() {
+            // Wait until block commit task is ready to write non-finalized blocks before dequeuing them
             self.send_ready_non_finalized_queued(parent_hash);
 
             let finalized_tip_height = self.read_service.db.finalized_tip_height().expect(
-                "Finalized state must have at least one block before committing non-finalized state",
+            "Finalized state must have at least one block before committing non-finalized state",
-            );
+        );
 
             self.queued_non_finalized_blocks
                 .prune_by_height(finalized_tip_height);
@@ -1063,6 +1063,28 @@ impl Service<Request> for StateService {
                 .boxed()
             }
 
+            // Used by sync, inbound, and block verifier to check if a block is already in the state
+            // before downloading or validating it.
+            Request::KnownBlock(hash) => {
+                let timer = CodeTimer::start();
+
+                let read_service = self.read_service.clone();
+
+                async move {
+                    let response = read::non_finalized_state_contains_block_hash(
+                        &read_service.latest_non_finalized_state(),
+                        hash,
+                    )
+                    .or_else(|| read::finalized_state_contains_block_hash(&read_service.db, hash));
+
+                    // The work is done in the future.
+                    timer.finish(module_path!(), line!(), "Request::KnownBlock");
+
+                    Ok(Response::KnownBlock(response))
+                }
+                .boxed()
+            }
+
             // Runs concurrently using the ReadStateService
             Request::Tip
             | Request::Depth(_)

zebra-state/src/service/non_finalized_state/chain.rs

@@ -420,6 +420,12 @@ impl Chain {
         self.height_by_hash.get(&hash).cloned()
     }
 
+    /// Returns true is the chain contains the given block hash.
+    /// Returns false otherwise.
+    pub fn contains_block_hash(&self, hash: &block::Hash) -> bool {
+        self.height_by_hash.contains_key(hash)
+    }
+
     /// Returns the non-finalized tip block height and hash.
     pub fn non_finalized_tip(&self) -> (Height, block::Hash) {
         (

zebra-state/src/service/queued_blocks.rs

@@ -267,6 +267,9 @@ impl SentHashes {
     /// Used for finalized blocks close to the final checkpoint, so non-finalized blocks can look up
     /// their UTXOs.
     ///
+    /// Assumes that blocks are added in the order of their height between `finish_batch` calls
+    /// for efficient pruning.
+    ///
     /// For more details see `add()`.
     pub fn add_finalized(&mut self, block: &FinalizedBlock) {
         // Track known UTXOs in sent blocks.

zebra-state/src/service/read.rs

@@ -34,8 +34,9 @@ pub use block::{
     any_utxo, block, block_header, transaction, transaction_hashes_for_block, unspent_utxo, utxo,
 };
 pub use find::{
-    best_tip, block_locator, chain_contains_hash, depth, find_chain_hashes, find_chain_headers,
+    best_tip, block_locator, chain_contains_hash, depth, finalized_state_contains_block_hash,
-    hash_by_height, height_by_hash, next_median_time_past, tip, tip_height,
+    find_chain_hashes, find_chain_headers, hash_by_height, height_by_hash, next_median_time_past,
+    non_finalized_state_contains_block_hash, tip, tip_height,
 };
 pub use tree::{orchard_tree, sapling_tree};
 

zebra-state/src/service/read/find.rs

@@ -32,7 +32,7 @@ use crate::{
         non_finalized_state::{Chain, NonFinalizedState},
         read::{self, block::block_header, FINALIZED_STATE_QUERY_RETRIES},
     },
-    BoxError,
+    BoxError, KnownBlock,
 };
 
 #[cfg(test)]
@@ -101,6 +101,31 @@ where
     Some(tip.0 - height.0)
 }
 
+/// Returns the location of the block if present in the non-finalized state.
+/// Returns None if the block hash is not found in the non-finalized state.
+pub fn non_finalized_state_contains_block_hash(
+    non_finalized_state: &NonFinalizedState,
+    hash: block::Hash,
+) -> Option<KnownBlock> {
+    let mut chains_iter = non_finalized_state.chain_set.iter().rev();
+    let is_hash_in_chain = |chain: &Arc<Chain>| chain.contains_block_hash(&hash);
+
+    // Equivalent to `chain_set.iter().next_back()` in `NonFinalizedState.best_chain()` method.
+    let best_chain = chains_iter.next();
+
+    match best_chain.map(is_hash_in_chain) {
+        Some(true) => Some(KnownBlock::BestChain),
+        Some(false) if chains_iter.any(is_hash_in_chain) => Some(KnownBlock::SideChain),
+        Some(false) | None => None,
+    }
+}
+
+/// Returns the location of the block if present in the finalized state.
+/// Returns None if the block hash is not found in the finalized state.
+pub fn finalized_state_contains_block_hash(db: &ZebraDb, hash: block::Hash) -> Option<KnownBlock> {
+    db.contains_hash(hash).then_some(KnownBlock::BestChain)
+}
+
 /// Return the height for the block at `hash`, if `hash` is in `chain` or `db`.
 pub fn height_by_hash<C>(chain: Option<C>, db: &ZebraDb, hash: block::Hash) -> Option<Height>
 where

zebrad/src/components/inbound/downloads.rs

@@ -242,11 +242,9 @@ where
 
         let fut = async move {
             // Check if the block is already in the state.
-            // BUG: check if the hash is in any chain (#862).
+            match state.oneshot(zs::Request::KnownBlock(hash)).await {
-            // Depth only checks the main chain.
+                Ok(zs::Response::KnownBlock(None)) => Ok(()),
-            match state.oneshot(zs::Request::Depth(hash)).await {
+                Ok(zs::Response::KnownBlock(Some(_))) => Err("already present".into()),
-                Ok(zs::Response::Depth(None)) => Ok(()),
-                Ok(zs::Response::Depth(Some(_))) => Err("already present".into()),
                 Ok(_) => unreachable!("wrong response"),
                 Err(e) => Err(e),
             }?;

zebrad/src/components/sync.rs

@@ -1057,22 +1057,18 @@ where
 
     /// Returns `true` if the hash is present in the state, and `false`
     /// if the hash is not present in the state.
-    ///
-    /// TODO BUG: check if the hash is in any chain (#862)
-    /// Depth only checks the main chain.
     async fn state_contains(&mut self, hash: block::Hash) -> Result<bool, Report> {
         match self
             .state
             .ready()
             .await
             .map_err(|e| eyre!(e))?
-            .call(zebra_state::Request::Depth(hash))
+            .call(zebra_state::Request::KnownBlock(hash))
             .await
             .map_err(|e| eyre!(e))?
         {
-            zs::Response::Depth(Some(_)) => Ok(true),
+            zs::Response::KnownBlock(loc) => Ok(loc.is_some()),
-            zs::Response::Depth(None) => Ok(false),
+            _ => unreachable!("wrong response to known block request"),
-            _ => unreachable!("wrong response to depth request"),
         }
     }
 

zebrad/src/components/sync/tests/timing.rs

@@ -144,11 +144,11 @@ fn request_genesis_is_rate_limited() {
     // panic in any other type of request.
     let state_service = tower::service_fn(move |request| {
         match request {
-            zebra_state::Request::Depth(_) => {
+            zebra_state::Request::KnownBlock(_) => {
                 // Track the call
                 state_requests_counter_in_service.fetch_add(1, Ordering::SeqCst);
                 // Respond with `None`
-                future::ok(zebra_state::Response::Depth(None))
+                future::ok(zebra_state::Response::KnownBlock(None))
             }
             _ => unreachable!("no other request is allowed"),
         }

zebrad/src/components/sync/tests/vectors.rs

@@ -78,9 +78,9 @@ async fn sync_blocks_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Block 0 is fetched and committed to the state
     peer_set
@@ -100,9 +100,9 @@ async fn sync_blocks_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis again
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(Some(0)));
+        .respond(zs::Response::KnownBlock(Some(zs::KnownBlock::BestChain)));
 
     // ChainSync::obtain_tips
 
@@ -127,9 +127,9 @@ async fn sync_blocks_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for the first unknown block (block 1)
     state_service
-        .expect_request(zs::Request::Depth(block1_hash))
+        .expect_request(zs::Request::KnownBlock(block1_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Clear remaining block locator requests
     for _ in 0..(sync::FANOUT - 1) {
@@ -148,13 +148,13 @@ async fn sync_blocks_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for all non-tip blocks (blocks 1 & 2) in response order
     state_service
-        .expect_request(zs::Request::Depth(block1_hash))
+        .expect_request(zs::Request::KnownBlock(block1_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
     state_service
-        .expect_request(zs::Request::Depth(block2_hash))
+        .expect_request(zs::Request::KnownBlock(block2_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Blocks 1 & 2 are fetched in order, then verified concurrently
     peer_set
@@ -305,9 +305,9 @@ async fn sync_blocks_duplicate_hashes_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Block 0 is fetched and committed to the state
     peer_set
@@ -327,9 +327,9 @@ async fn sync_blocks_duplicate_hashes_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis again
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(Some(0)));
+        .respond(zs::Response::KnownBlock(Some(zs::KnownBlock::BestChain)));
 
     // ChainSync::obtain_tips
 
@@ -356,9 +356,9 @@ async fn sync_blocks_duplicate_hashes_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for the first unknown block (block 1)
     state_service
-        .expect_request(zs::Request::Depth(block1_hash))
+        .expect_request(zs::Request::KnownBlock(block1_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Clear remaining block locator requests
     for _ in 0..(sync::FANOUT - 1) {
@@ -377,13 +377,13 @@ async fn sync_blocks_duplicate_hashes_ok() -> Result<(), crate::BoxError> {
 
     // State is checked for all non-tip blocks (blocks 1 & 2) in response order
     state_service
-        .expect_request(zs::Request::Depth(block1_hash))
+        .expect_request(zs::Request::KnownBlock(block1_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
     state_service
-        .expect_request(zs::Request::Depth(block2_hash))
+        .expect_request(zs::Request::KnownBlock(block2_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Blocks 1 & 2 are fetched in order, then verified concurrently
     peer_set
@@ -520,9 +520,9 @@ async fn sync_block_lookahead_drop() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Block 0 is fetched, but the peer returns a much higher block.
     // (Mismatching hashes are usually ignored by the network service,
@@ -587,9 +587,9 @@ async fn sync_block_too_high_obtain_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Block 0 is fetched and committed to the state
     peer_set
@@ -609,9 +609,9 @@ async fn sync_block_too_high_obtain_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis again
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(Some(0)));
+        .respond(zs::Response::KnownBlock(Some(zs::KnownBlock::BestChain)));
 
     // ChainSync::obtain_tips
 
@@ -637,9 +637,9 @@ async fn sync_block_too_high_obtain_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for the first unknown block (block 982k)
     state_service
-        .expect_request(zs::Request::Depth(block982k_hash))
+        .expect_request(zs::Request::KnownBlock(block982k_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Clear remaining block locator requests
     for _ in 0..(sync::FANOUT - 1) {
@@ -658,17 +658,17 @@ async fn sync_block_too_high_obtain_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for all non-tip blocks (blocks 982k, 1, 2) in response order
     state_service
-        .expect_request(zs::Request::Depth(block982k_hash))
+        .expect_request(zs::Request::KnownBlock(block982k_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
     state_service
-        .expect_request(zs::Request::Depth(block1_hash))
+        .expect_request(zs::Request::KnownBlock(block1_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
     state_service
-        .expect_request(zs::Request::Depth(block2_hash))
+        .expect_request(zs::Request::KnownBlock(block2_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Blocks 982k, 1, 2 are fetched in order, then verified concurrently,
     // but block 982k verification is skipped because it is too high.
@@ -748,9 +748,9 @@ async fn sync_block_too_high_extend_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Block 0 is fetched and committed to the state
     peer_set
@@ -770,9 +770,9 @@ async fn sync_block_too_high_extend_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for genesis again
     state_service
-        .expect_request(zs::Request::Depth(block0_hash))
+        .expect_request(zs::Request::KnownBlock(block0_hash))
         .await
-        .respond(zs::Response::Depth(Some(0)));
+        .respond(zs::Response::KnownBlock(Some(zs::KnownBlock::BestChain)));
 
     // ChainSync::obtain_tips
 
@@ -797,9 +797,9 @@ async fn sync_block_too_high_extend_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for the first unknown block (block 1)
     state_service
-        .expect_request(zs::Request::Depth(block1_hash))
+        .expect_request(zs::Request::KnownBlock(block1_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Clear remaining block locator requests
     for _ in 0..(sync::FANOUT - 1) {
@@ -818,13 +818,13 @@ async fn sync_block_too_high_extend_tips() -> Result<(), crate::BoxError> {
 
     // State is checked for all non-tip blocks (blocks 1 & 2) in response order
     state_service
-        .expect_request(zs::Request::Depth(block1_hash))
+        .expect_request(zs::Request::KnownBlock(block1_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
     state_service
-        .expect_request(zs::Request::Depth(block2_hash))
+        .expect_request(zs::Request::KnownBlock(block2_hash))
         .await
-        .respond(zs::Response::Depth(None));
+        .respond(zs::Response::KnownBlock(None));
 
     // Blocks 1 & 2 are fetched in order, then verified concurrently
     peer_set