fix: bitcoin adapter: prune blocks if response empty (#3775)

Before, blocks were only pruned if they were: either processed by the
canister, or below (less height) the anchor.

After, in addition to the above, all blocks are also pruned if the
response contained no blocks. The reason why this is safe is because
during the BFS traversal, we try to return at least one cached block
that is reachable from the anchor. If no block was found, this indicates
that no block in the cache can be reached from the anchor, so they can
be safely discarded.

Secondly, this is also needed because the cache can get full of blocks
which are above the height of the anchor, but can't be served currently
(there is for example a block that's missing between them and the
anchor). And if the cache is full of blocks, the adapter won't be able
to request any more new blocks.

Author: mihailjianu1

rs/bitcoin/adapter/src/blockchainmanager.rs

@@ -42,13 +42,6 @@ const MAX_UNSOLICITED_HEADERS: usize = 20;
 /// to a peer at a time.
 const INV_PER_GET_DATA_REQUEST: u32 = 8;
 
-const ONE_MB: usize = 1_024 * 1_024;
-
-/// The limit at which we should stop making additional requests for new blocks as the block cache
-/// becomes too large. Inflight `getdata` messages will remain active, but new `getdata` messages will
-/// not be created.
-const BLOCK_CACHE_THRESHOLD_BYTES: usize = 10 * ONE_MB;
-
 /// Block locators. Consists of starting hashes and a stop hash.
 type Locators = (Vec<BlockHash>, BlockHash);
 
@@ -540,17 +533,12 @@ impl BlockchainManager {
             return;
         }
 
-        let block_cache_size = self.blockchain.lock().unwrap().get_block_cache_size();
+        let is_cache_full = self.blockchain.lock().unwrap().is_block_cache_full();
 
-        if block_cache_size >= BLOCK_CACHE_THRESHOLD_BYTES {
-            debug!(
-                self.logger,
-                "Cache Size: {}, Max Size: {}", block_cache_size, BLOCK_CACHE_THRESHOLD_BYTES
-            );
+        if is_cache_full {
+            debug!(self.logger, "Cache full");
         }
 
-        let is_cache_full = block_cache_size >= BLOCK_CACHE_THRESHOLD_BYTES;
-
         // Count the number of requests per peer.
         let mut requests_per_peer: HashMap<SocketAddr, u32> =
             self.peer_info.keys().map(|addr| (*addr, 0)).collect();
@@ -1302,7 +1290,7 @@ pub mod test {
                 blockchain.add_block(block).expect("failed to add block");
             }
 
-            assert!(blockchain.get_block_cache_size() >= BLOCK_CACHE_THRESHOLD_BYTES);
+            assert!(blockchain.is_block_cache_full());
         }
 
         let block_1_hash = large_blockchain_headers

rs/bitcoin/adapter/src/blockchainstate.rs

@@ -12,6 +12,12 @@ use thiserror::Error;
 
 use bitcoin::Work;
 
+/// The limit at which we should stop making additional requests for new blocks as the block cache
+/// becomes too large. Inflight `getdata` messages will remain active, but new `getdata` messages will
+/// not be created.
+const BLOCK_CACHE_THRESHOLD_BYTES: usize = 10 * ONE_MB;
+const ONE_MB: usize = 1_024 * 1_024;
+
 /// Contains the necessary information about a tip.
 #[derive(Clone, Debug)]
 pub struct Tip {
@@ -137,6 +143,11 @@ impl BlockchainState {
         self.header_cache.get(hash)
     }
 
+    /// Returns the hashes of all cached blocks.
+    pub(crate) fn get_cached_blocks(&self) -> Vec<BlockHash> {
+        self.block_cache.keys().copied().collect()
+    }
+
     /// Processes the `headers` message received from Bitcoin nodes by adding them to the state.
     /// Headers are expected to be sorted. If they are not, the headers will be likely be rejected
     /// with a [AddHeaderError::PrevHeaderNotCached](AddHeaderError::PrevHeaderNotCached) error.
@@ -328,6 +339,10 @@ impl BlockchainState {
         self.block_cache = HashMap::new();
     }
 
+    pub(crate) fn is_block_cache_full(&self) -> bool {
+        self.get_block_cache_size() >= BLOCK_CACHE_THRESHOLD_BYTES
+    }
+
     /// Returns the current size of the block cache.
     pub fn get_block_cache_size(&self) -> usize {
         self.block_cache

rs/bitcoin/adapter/src/get_successors_handler.rs

@@ -118,7 +118,7 @@ impl GetSuccessorsHandler {
             .processed_block_hashes
             .observe(request.processed_block_hashes.len() as f64);
 
-        let (blocks, next) = {
+        let (blocks, next, obsolete_blocks) = {
             let state = self.state.lock().unwrap();
             let anchor_height = state
                 .get_cached_header(&request.anchor)
@@ -139,7 +139,17 @@ impl GetSuccessorsHandler {
                 &blocks,
                 self.network,
             );
-            (blocks, next)
+            // If no blocks are returned, this means that nothing that is in the cache could be reached from the anchor.
+            // We can safely remove everything that is in the cache then, as those blocks are no longer needed.
+            // There is a chance that these blocks are above the anchor height (but they were forked below it),
+            // meaning that the regular "pruning anything below anchor" will not affect them.
+            // There is also a chance that they are reachable from the anchor, just not through the cache.
+            // Meaning that we still need to download some other blocks first. (hence we need to free the cache).
+            let mut obsolete_blocks = request.processed_block_hashes;
+            if blocks.is_empty() && state.is_block_cache_full() {
+                obsolete_blocks.extend(state.get_cached_blocks())
+            }
+            (blocks, next, obsolete_blocks)
         };
         let response_next = &next[..next.len().min(MAX_NEXT_BLOCK_HEADERS_LENGTH)];
         let response = GetSuccessorsResponse {
@@ -160,7 +170,7 @@ impl GetSuccessorsHandler {
         self.blockchain_manager_tx
             .try_send(BlockchainManagerRequest::PruneBlocks(
                 request.anchor,
-                request.processed_block_hashes,
+                obsolete_blocks,
             ))
             .ok();
 
@@ -195,30 +205,31 @@ fn get_successor_blocks(
         _ => MAX_BLOCKS_BYTES,
     };
 
+    let max_blocks_length = if allow_multiple_blocks {
+        MAX_BLOCKS_LENGTH
+    } else {
+        1
+    };
+
     // Compute the blocks by starting a breadth-first search.
     while let Some(block_hash) = queue.pop_front() {
         if !seen.contains(block_hash) {
             // Retrieve the block from the cache.
-            match state.get_block(block_hash) {
-                Some(block) => {
-                    let block_size = block.total_size();
-                    if response_block_size == 0
-                        || (response_block_size + block_size <= max_blocks_size
-                            && successor_blocks.len() < MAX_BLOCKS_LENGTH
-                            && allow_multiple_blocks)
-                    {
-                        successor_blocks.push(block.clone());
-                        response_block_size += block_size;
-                    } else {
-                        break;
-                    }
-                }
-                None => {
-                    // Cache miss has occurred. This block or any of its successors cannot
-                    // be returned. Discarding this subtree from the BFS.
-                    continue;
-                }
+            let Some(block) = state.get_block(block_hash) else {
+                // If the block is not in the cache, we skip it and all its subtree.
+                // We don't want to return orphaned blocks to the canister.
+                continue;
+            };
+            let block_size = block.total_size();
+            // If we have at least one block in the response, and we can't fit another block, we stop.
+            if response_block_size > 0
+                && (response_block_size + block_size > max_blocks_size
+                    || successor_blocks.len() + 1 > max_blocks_length)
+            {
+                break;
             }
+            successor_blocks.push(block);
+            response_block_size += block_size;
         }
 
         queue.extend(

rs/bitcoin/adapter/tests/adapter_test.rs

@@ -830,19 +830,19 @@ fn test_receives_blocks_from_forks() {
         .get_new_address(None, None)
         .unwrap()
         .assume_checked();
-    client1.generate_to_address(25, &address1).unwrap();
+    client1.generate_to_address(10, &address1).unwrap();
 
-    wait_for_blocks(&client1, 25);
-    wait_for_blocks(&client2, 25);
+    wait_for_blocks(&client1, 10);
+    wait_for_blocks(&client2, 10);
 
     let address2 = client2
         .get_new_address(None, None)
         .unwrap()
         .assume_checked();
-    client2.generate_to_address(25, &address2).unwrap();
+    client2.generate_to_address(10, &address2).unwrap();
 
-    wait_for_blocks(&client1, 50);
-    wait_for_blocks(&client2, 50);
+    wait_for_blocks(&client1, 20);
+    wait_for_blocks(&client2, 20);
 
     // Disconnect the nodes to create a fork
     client1
@@ -852,15 +852,15 @@ fn test_receives_blocks_from_forks() {
     wait_for_connection(&client1, 1);
     wait_for_connection(&client2, 1);
 
-    client1.generate_to_address(10, &address1).unwrap();
-    client2.generate_to_address(15, &address2).unwrap();
+    client1.generate_to_address(3, &address1).unwrap();
+    client2.generate_to_address(6, &address2).unwrap();
 
-    wait_for_blocks(&client1, 60);
-    wait_for_blocks(&client2, 65);
+    wait_for_blocks(&client1, 23);
+    wait_for_blocks(&client2, 26);
 
     let anchor = client1.get_block_hash(0).unwrap()[..].to_vec();
-    let blocks = sync_blocks(&adapter_client, &mut vec![], anchor, 75, 200);
-    assert_eq!(blocks.len(), 75);
+    let blocks = sync_blocks(&adapter_client, &mut vec![], anchor, 29, 201);
+    assert_eq!(blocks.len(), 29);
 }
 
 /// Checks that the adapter returns blocks in BFS order.