Chainindex passthrough when behind backing validator

integration-tests/tests/chain_cache.rs

@@ -306,6 +306,7 @@ mod chain_query_interface {
                 block_hash,
                 &indexer
                     .find_fork_point(&snapshot, block_hash)
+                    .await
                     .unwrap()
                     .unwrap()
                     .0

zaino-state/src/chain_index.rs

@@ -12,6 +12,7 @@
 //!   - NOTE: Full transaction and block data is served from the backend finalizer.
 
 use crate::chain_index::non_finalised_state::BestTip;
+use crate::chain_index::source::GetTransactionLocation;
 use crate::chain_index::types::{BestChainLocation, NonBestChainLocation};
 use crate::error::{ChainIndexError, ChainIndexErrorKind, FinalisedStateError};
 use crate::IndexedBlock;
@@ -188,6 +189,7 @@ pub trait ChainIndex {
     /// between the given heights.
     /// Returns None if the specified end height
     /// is greater than the snapshot's tip
+    // TO-TEST
     #[allow(clippy::type_complexity)]
     fn get_block_range(
         &self,
@@ -202,7 +204,7 @@ pub trait ChainIndex {
         &self,
         snapshot: &Self::Snapshot,
         block_hash: &types::BlockHash,
-    ) -> Result<Option<(types::BlockHash, types::Height)>, Self::Error>;
+    ) -> impl std::future::Future<Output = Result<Option<(types::BlockHash, types::Height)>, Self::Error>>;
 
     /// Returns the block commitment tree data by hash
     #[allow(clippy::type_complexity)]
@@ -225,10 +227,10 @@ pub trait ChainIndex {
     ) -> impl std::future::Future<Output = Result<Option<(Vec<u8>, Option<u32>)>, Self::Error>>;
 
     /// Given a transaction ID, returns all known hashes and heights of blocks
-    /// containing that transaction. Height is None for blocks not on the best chain.
+    /// containing that transaction.
     ///
-    /// Also returns a bool representing whether the transaction is *currently* in the mempool.
-    /// This is not currently tied to the given snapshot but rather uses the live mempool.
+    /// Also returns if the transaction is in the mempool (and whether that mempool is
+    /// in-sync with the provided snapshot)
     #[allow(clippy::type_complexity)]
     fn get_transaction_status(
         &self,
@@ -576,6 +578,25 @@ impl<Source: BlockchainSource> NodeBackedChainIndexSubscriber<Source> {
             .transpose()
     }
 
+    async fn get_snapshot_block_height(
+        &self,
+        nonfinalized_snapshot: &NonfinalizedBlockCacheSnapshot,
+        hash: types::BlockHash,
+    ) -> Result<Option<types::Height>, ChainIndexError> {
+        match nonfinalized_snapshot.blocks.get(&hash).cloned() {
+            Some(block) => Ok(nonfinalized_snapshot
+                .heights_to_hashes
+                .values()
+                .find(|h| **h == hash)
+                // Canonical height is None for blocks not on the best chain
+                .map(|_| block.index().height())),
+            None => match self.finalized_state.get_block_height(hash).await {
+                Ok(height) => Ok(height),
+                Err(_e) => Err(ChainIndexError::database_hole(hash)),
+            },
+        }
+    }
+
     /**
     Searches finalized and non-finalized chains for any blocks containing the transaction.
     Ordered with finalized blocks first.
@@ -618,6 +639,49 @@ impl<Source: BlockchainSource> NodeBackedChainIndexSubscriber<Source> {
         Ok(finalized_blocks_containing_transaction
             .chain(non_finalized_blocks_containing_transaction))
     }
+
+    async fn get_block_height_passthrough(
+        &self,
+        nonfinalized_snapshot: &NonfinalizedBlockCacheSnapshot,
+        hash: types::BlockHash,
+    ) -> Result<Option<types::Height>, ChainIndexError> {
+        match self
+            .blockchain_source
+            .get_block(HashOrHeight::Hash(hash.into()))
+            .await
+        {
+            Ok(Some(block))
+                if types::Height::from(block.coinbase_height().expect("block to have height"))
+                    <= nonfinalized_snapshot.validator_finalized_height =>
+            {
+                Ok(Some(block.coinbase_height().unwrap().into()))
+            }
+            Ok(_) => Ok(None),
+            Err(e) => Err(ChainIndexError::backing_validator(e)),
+        }
+    }
+
+    // Get the height of the mempool
+    fn get_mempool_height(
+        &self,
+        snapshot: &NonfinalizedBlockCacheSnapshot,
+    ) -> Option<types::Height> {
+        snapshot
+            .blocks
+            .iter()
+            .find(|(hash, _block)| **hash == self.mempool.mempool_chain_tip())
+            .map(|(_hash, block)| block.height())
+    }
+
+    fn mempool_branch_id(&self, snapshot: &NonfinalizedBlockCacheSnapshot) -> Option<u32> {
+        self.get_mempool_height(snapshot).and_then(|height| {
+            ConsensusBranchId::current(
+                &self.non_finalized_state.network,
+                zebra_chain::block::Height::from(height + 1),
+            )
+            .map(u32::from)
+        })
+    }
 }
 
 impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Source> {
@@ -641,32 +705,37 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
         nonfinalized_snapshot: &Self::Snapshot,
         hash: types::BlockHash,
     ) -> Result<Option<types::Height>, Self::Error> {
-        match nonfinalized_snapshot.blocks.get(&hash).cloned() {
-            Some(block) => Ok(nonfinalized_snapshot
-                .heights_to_hashes
-                .values()
-                .find(|h| **h == hash)
-                // Canonical height is None for blocks not on the best chain
-                .map(|_| block.index().height())),
-            None => match self.finalized_state.get_block_height(hash).await {
-                Ok(height) => Ok(height),
-                Err(_e) => Err(ChainIndexError::database_hole(hash)),
-            },
+        let snapshot_block_height = self
+            .get_snapshot_block_height(nonfinalized_snapshot, hash)
+            .await?;
+        match snapshot_block_height {
+            Some(h) => Ok(Some(h)),
+            None => {
+                self.get_block_height_passthrough(nonfinalized_snapshot, hash)
+                    .await
+            }
         }
     }
 
     /// Given inclusive start and end heights, stream all blocks
     /// between the given heights.
-    /// Returns None if the specified end height
-    /// is greater than the snapshot's tip
+    /// Returns None if the specified start height
+    /// is greater than the snapshot's tip and greater
+    /// than the validator's finalized height (100 blocks below tip)
     fn get_block_range(
         &self,
         nonfinalized_snapshot: &Self::Snapshot,
         start: types::Height,
         end: std::option::Option<types::Height>,
     ) -> Option<impl Stream<Item = Result<Vec<u8>, Self::Error>>> {
-        let end = end.unwrap_or(nonfinalized_snapshot.best_tip.height);
-        if end <= nonfinalized_snapshot.best_tip.height {
+        // We can serve blocks above where the validator has finalized
+        // only if we have those blocks in our nonfinalized snapshot
+        let max_servable_height = nonfinalized_snapshot
+            .validator_finalized_height
+            .max(nonfinalized_snapshot.best_tip.height);
+        let end = end.unwrap_or(max_servable_height);
+        // Serve as high as we can, or to the provided end if it's lower
+        if start <= max_servable_height.min(end) {
             Some(
                 futures::stream::iter((start.0)..=(end.0)).then(move |height| async move {
                     match self
@@ -697,7 +766,12 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
                                         .await?
                                         .ok_or(ChainIndexError::database_hole(block.hash()))
                                 }
-                                None => Err(ChainIndexError::database_hole(height)),
+                                None => self
+                                    .get_fullblock_bytes_from_node(HashOrHeight::Height(
+                                        zebra_chain::block::Height(height),
+                                    ))
+                                    .await?
+                                    .ok_or(ChainIndexError::database_hole(height)),
                             }
                         }
                     }
@@ -710,30 +784,62 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
 
     /// Finds the newest ancestor of the given block on the main
     /// chain, or the block itself if it is on the main chain.
-    fn find_fork_point(
+    /// Returns Ok(None) if no fork point found. This is not an error,
+    /// as zaino does not guarentee knowledge of all sidechain data.
+    async fn find_fork_point(
         &self,
         snapshot: &Self::Snapshot,
         block_hash: &types::BlockHash,
     ) -> Result<Option<(types::BlockHash, types::Height)>, Self::Error> {
         let Some(block) = snapshot.as_ref().get_chainblock_by_hash(block_hash) else {
-            // No fork point found. This is not an error,
-            // as zaino does not guarentee knowledge of all sidechain data.
-            return Ok(None);
+            // We don't have the block in our non-finalized state,
+            // we'll only be aware of it if it's main-chain.
+            // Find it from the source, and return its height and hash
+            return match self
+                .blockchain_source
+                .get_block(HashOrHeight::Hash(zebra_chain::block::Hash::from(
+                    *block_hash,
+                )))
+                .await
+            {
+                Ok(Some(block))
+                    // We don't have the block in our non-finalized state
+                    // we can only passthrough assuming the block is finalized
+                    if block.coinbase_height().unwrap()
+                        <= snapshot.validator_finalized_height =>
+                {
+                    Ok(Some((
+                        types::BlockHash::from(block.hash()),
+                        types::Height::from(block.coinbase_height().unwrap()),
+                    )))
+                }
+                // The block is non-finalized, and we haven't synced it yet.
+                // We can't make any assertions about the best chain
+                // if it's not in our snapshot.
+                // TODO: Should this be an error?
+                Ok(_) => Ok(None),
+                Err(e) => Err(ChainIndexError::backing_validator(e)),
+            };
         };
+        // If we have the block in our heights_to_hashes set, it's main-chain
+        // Return it's hash and height
         if snapshot.heights_to_hashes.get(&block.height()) == Some(block.hash()) {
             Ok(Some((*block.hash(), block.height())))
+            // Otherwise, it's non-best chain! Grab its parent, and recurse
         } else {
-            self.find_fork_point(snapshot, block.index().parent_hash())
+            // gotta pin recursive async functions to prevent infinite-sized
+            // Future-implementing types
+            Box::pin(self.find_fork_point(snapshot, block.index().parent_hash())).await
         }
     }
 
     /// Returns the block commitment tree data by hash
     async fn get_treestate(
         &self,
-        // snapshot: &Self::Snapshot,
         // currently not implemented internally, fetches data from validator.
-        //
-        // NOTE: Should this check blockhash exists in snapshot and db before proxying call?
+        // as this looks up the block by hash, and cares not if the
+        // block is on the main chain or not, this is safe to pass through
+        // even if the target block is non-finalized
         hash: &types::BlockHash,
     ) -> Result<(Option<Vec<u8>>, Option<Vec<u8>>), Self::Error> {
         match self.blockchain_source.get_treestate(*hash).await {
@@ -747,6 +853,8 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
     }
 
     /// given a transaction id, returns the transaction
+    /// and the consensus branch ID for the block the transaction
+    /// is in
     async fn get_raw_transaction(
         &self,
         snapshot: &Self::Snapshot,
@@ -760,22 +868,34 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
             .await
         {
             let bytes = mempool_tx.serialized_tx.as_ref().as_ref().to_vec();
-            let mempool_height = snapshot
-                .blocks
-                .iter()
-                .find(|(hash, _block)| **hash == self.mempool.mempool_chain_tip())
-                .map(|(_hash, block)| block.height());
-            let mempool_branch_id = mempool_height.and_then(|height| {
-                ConsensusBranchId::current(
-                    &self.non_finalized_state.network,
-                    zebra_chain::block::Height::from(height + 1),
-                )
-                .map(u32::from)
-            });
+            let mempool_branch_id = self.mempool_branch_id(snapshot);
 
             return Ok(Some((bytes, mempool_branch_id)));
         }
 
+        if let Some((transaction, location)) = self
+            .blockchain_source
+            .get_transaction(*txid)
+            .await
+            .map_err(|e| ChainIndexError::backing_validator(e))?
+        {
+            // Passthrough, if the transaction is finalized
+            // on the best chain
+            if let source::GetTransactionLocation::BestChain(height) = location {
+                if height <= snapshot.validator_finalized_height {
+                    return Ok(Some((
+                        zebra_chain::transaction::SerializedTransaction::from(transaction)
+                            .as_ref()
+                            .to_vec(),
+                        ConsensusBranchId::current(&self.non_finalized_state.network, height)
+                            .map(u32::from),
+                    )));
+                }
+            }
+        }
+
+        // if the tranasction isn't finalized on the best chain
+        // check our indexes
         let Some(block) = self
             .blocks_containing_transaction(snapshot, txid.0)
             .await?
@@ -784,13 +904,6 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
             return Ok(None);
         };
 
-        // NOTE: Could we safely use zebra's get transaction method here without invalidating the snapshot?
-        // This would be a more efficient way to fetch transaction data.
-        //
-        // Should NodeBackedChainIndex keep a clone of source to use here?
-        //
-        // This will require careful attention as there is a case where a transaction may still exist,
-        // but may have been reorged into a different block, possibly breaking the validation of this interface.
         let full_block = self
             .non_finalized_state
             .source
@@ -888,6 +1001,31 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
             }
         }
 
+        // If we haven't found a block on the best chain,
+        // try passthrough
+        if best_chain_block == None {
+            if let Some((_transaction, location)) = self
+                .blockchain_source
+                .get_transaction(*txid)
+                .await
+                .map_err(|e| ChainIndexError::backing_validator(e))?
+            {
+                if let GetTransactionLocation::BestChain(height) = location {
+                    if height <= snapshot.validator_finalized_height {
+                        if let Some(block) = self
+                            .blockchain_source
+                            .get_block(HashOrHeight::Height(height))
+                            .await
+                            .map_err(|e| ChainIndexError::backing_validator(e))?
+                        {
+                            best_chain_block =
+                                Some(BestChainLocation::Block(block.hash().into(), height.into()));
+                        }
+                    }
+                }
+            }
+        }
+
         Ok((best_chain_block, non_best_chain_blocks))
     }
 

zaino-state/src/chain_index/mempool.rs

@@ -246,7 +246,7 @@ impl<T: BlockchainSource> Mempool<T> {
         })?;
 
         for txid in txids {
-            let transaction = self
+            let (transaction, _location) = self
                 .fetcher
                 .get_transaction(txid.0.into())
                 .await?