handle missing blocks from send_blocks_proof.rs

Signed-off-by: Eval EXEC <execvy@gmail.com>

Author: eval-exec

light-client-lib/src/protocols/filter/block_filter.rs

@@ -151,18 +151,71 @@ impl FilterProtocol {
                     db_matched_blocks.blocks_count,
                     matched_blocks.len(),
                 );
+                db_matched_blocks.blocks.iter().for_each(|b| {
+                    debug!("db matched block: {}, proved={}", b.hash, b.proved);
+                });
                 let option = matched_blocks.is_empty();
 
                 if option {
-                    // recover matched blocks from storage
-                    self.peers.add_matched_blocks(
-                        &mut matched_blocks,
-                        db_matched_blocks
-                            .blocks
-                            .into_iter()
-                            .map(|b| (b.hash, b.proved))
-                            .collect(),
-                    );
+                    debug!("matched_blocks is empty");
+
+                    // Filter out blocks already marked as missing (e.g., uncle blocks)
+                    let filtered_blocks: Vec<_> = db_matched_blocks
+                        .blocks
+                        .iter()
+                        .filter(|b| {
+                            if let Some((_added_ts, _first_sent, missing)) =
+                                self.peers.get_header_fetch_info(&b.hash)
+                            {
+                                debug!(
+                                    "header fetch info for block {:#x}:added_ts={}, first_sent={}, missing={}",
+                                    b.hash,
+                                    _added_ts,
+                                    _first_sent,
+                                    missing
+                                );
+                                if missing {
+                                    debug!(
+                                        "Skipping block {:#x} from DB - marked as missing",
+                                        b.hash
+                                    );
+                                    return false;
+                                }
+                            } else {
+                                trace!("not found header fetch info for block {:#x}", b.hash);
+                            }
+                            true
+                        })
+                        .map(|b| (b.hash.clone(), b.proved))
+                        .collect();
+
+                    if filtered_blocks.is_empty() {
+                        // All blocks in this DB range are missing - remove it and load next
+                        info!(
+                            "All {} blocks in DB range {} are missing (uncle blocks), removing",
+                            db_matched_blocks.blocks.len(),
+                            db_matched_blocks.start_number
+                        );
+                        self.storage
+                            .remove_matched_blocks(db_matched_blocks.start_number);
+
+                        // Also remove these blocks from fetching_headers to prevent memory leak
+                        for block in &db_matched_blocks.blocks {
+                            self.peers.remove_fetching_header(&block.hash);
+                        }
+
+                        // Will load next range on next iteration
+                        return;
+                    } else {
+                        debug!(
+                            "add filtered blocks into matched_blocks, count={}",
+                            filtered_blocks.len()
+                        );
+                    }
+
+                    // recover matched blocks from storage (only non-missing ones)
+                    self.peers
+                        .add_matched_blocks(&mut matched_blocks, filtered_blocks);
                     let tip_header = self.storage.get_tip_header();
                     prove_or_download_matched_blocks(
                         Arc::clone(&self.peers),
@@ -178,6 +231,11 @@ impl FilterProtocol {
                         );
                         self.send_get_block_filters(nc, *peer, start_number);
                     }
+                } else {
+                    debug!("matched blocks are not empty:");
+                    matched_blocks.iter().for_each(|matched_block| {
+                        debug!("matched_block: {}, {:?}", matched_block.0, matched_block.1);
+                    });
                 }
             } else if self.should_ask(immediately).await && could_ask_more {
                 debug!(
@@ -266,10 +324,9 @@ impl FilterProtocol {
         peer: PeerIndex,
         start_number: BlockNumber,
     ) {
-        trace!(
+        debug!(
             "request block filter from peer {}, starts at {}",
-            peer,
-            start_number
+            peer, start_number
         );
         let content = packed::GetBlockFilters::new_builder()
             .start_number(start_number)

light-client-lib/src/protocols/filter/components/block_filter_hashes_process.rs

@@ -1,6 +1,6 @@
 use ckb_network::{BoxedCKBProtocolContext, PeerIndex};
 use ckb_types::{core::BlockNumber, packed, prelude::*};
-use log::trace;
+use log::{debug, trace};
 use rand::seq::SliceRandom as _;
 
 use crate::protocols::{FilterProtocol, Status, StatusCode};
@@ -28,6 +28,7 @@ impl<'a> BlockFilterHashesProcess<'a> {
     }
 
     pub async fn execute(self) -> Status {
+        debug!("process BlockFilterHashes message");
         let peer_state = if let Some(peer_state) = self.protocol.peers.get_state(&self.peer_index) {
             peer_state
         } else {
@@ -51,7 +52,7 @@ impl<'a> BlockFilterHashesProcess<'a> {
             .map(|item| item.to_entity())
             .collect::<Vec<_>>();
 
-        trace!(
+        debug!(
             "peer {}: last-state: {}, add block filter hashes (start: {}, len: {}) \
             and parent block filter hash is {:#x}",
             self.peer_index,

light-client-lib/src/protocols/filter/components/block_filters_process.rs

@@ -33,6 +33,8 @@ impl<'a> BlockFiltersProcess<'a> {
     }
 
     pub async fn execute(self) -> Status {
+        debug!("block filters process execute");
+
         if self.filter.storage.is_filter_scripts_empty() {
             info!("ignoring, filter scripts may have been cleared during syncing");
             return Status::ok();

light-client-lib/src/protocols/light_client/components/send_blocks_proof.rs

@@ -39,6 +39,7 @@ impl<'a> SendBlocksProofProcess<'a> {
 
     pub(crate) async fn execute(self) -> Status {
         let status = self.execute_internally().await;
+        debug!("block proof status: {}", status);
         self.protocol
             .peers()
             .update_blocks_proof_request(self.peer_index, None, false);
@@ -98,6 +99,9 @@ impl<'a> SendBlocksProofProcess<'a> {
             .to_entity()
             .into_iter()
             .collect::<Vec<_>>();
+
+        debug!("got block proof: missing {:?}", &missing_block_hashes);
+
         if !original_request.check_block_hashes(&received_block_hashes, &missing_block_hashes) {
             error!("peer {} send an unknown proof", self.peer_index);
             return StatusCode::UnexpectedResponse.into();
@@ -224,6 +228,52 @@ impl<'a> SendBlocksProofProcess<'a> {
         self.protocol
             .peers()
             .mark_fetching_headers_missing(&missing_block_hashes);
+
+        // Remove missing blocks from matched_blocks to prevent batch stall
+        // This is safe because:
+        // 1. If these are uncle blocks from an old fork, they've already been re-filtered
+        // 2. If from a recent reorg, SendLastStateProof will detect it and trigger
+        //    rollback_to_block(), which resets min_filtered_block_number and re-runs filters
+        // 3. New main chain blocks at the same heights will be checked during re-filtering
+        if original_request.should_get_blocks() && !missing_block_hashes.is_empty() {
+            let mut matched_blocks = self.protocol.peers().matched_blocks().write().await;
+            let mut removed_count = 0;
+
+            for missing_hash in &missing_block_hashes {
+                if matched_blocks.remove(&missing_hash.unpack()).is_some() {
+                    removed_count += 1;
+                    debug!(
+                        "Removed missing block {:#x} from matched_blocks \
+                         (likely uncle block or peer doesn't have it)",
+                        missing_hash
+                    );
+                }
+            }
+
+            if removed_count > 0 {
+                info!(
+                    "Removed {} missing block(s) from matched_blocks. \
+                     If due to reorg, filters will re-run from fork point to check new blocks.",
+                    removed_count
+                );
+
+                // Check if batch is now complete (all remaining blocks have been downloaded)
+                let all_downloaded = self
+                    .protocol
+                    .peers()
+                    .all_matched_blocks_downloaded(&matched_blocks);
+
+                if all_downloaded && !matched_blocks.is_empty() {
+                    info!(
+                        "Batch complete after removing missing blocks, {} blocks ready",
+                        matched_blocks.len()
+                    );
+                } else if matched_blocks.is_empty() {
+                    debug!("matched_blocks now empty after removing missing blocks");
+                }
+            }
+        }
+
         Status::ok()
     }
 }

light-client-lib/src/protocols/light_client/peers.rs

@@ -1216,9 +1216,23 @@ impl Peers {
     }
     pub(crate) fn mark_fetching_headers_missing(&self, block_hashes: &[Byte32]) {
         for block_hash in block_hashes {
-            if let Some(mut value) = self.fetching_headers.get_mut(block_hash) {
-                value.missing = true;
-            }
+            self.fetching_headers
+                .entry(block_hash.clone())
+                .and_modify(|info| {
+                    // Block was already in fetching_headers, mark as missing
+                    info.missing = true;
+                })
+                .or_insert_with(|| {
+                    // Block not in fetching_headers (e.g., loaded from matched_blocks DB)
+                    // Insert with missing=true to prevent infinite retry loop
+                    // This will be cleaned up when the DB range is removed
+                    FetchInfo {
+                        added_ts: unix_time_as_millis(),
+                        first_sent: 0,
+                        timeout: false,
+                        missing: true,
+                    }
+                });
         }
     }
     pub(crate) fn mark_fetching_txs_missing(&self, tx_hashes: &[Byte32]) {

light-client-lib/src/storage/db/native.rs

@@ -735,7 +735,6 @@ impl Storage {
             if entry.is_none() {
                 break;
             }
-            dbg!(&entry);
 
             let matched_blocks = entry.unwrap();
             let start_number = matched_blocks.start_number;

light-client-lib/src/utils/network.rs

@@ -13,6 +13,7 @@ pub(crate) fn prove_or_download_matched_blocks(
     nc: &dyn CKBProtocolContext,
     init_blocks_in_transit_per_peer: usize,
 ) {
+    info!("in prove_or_download-matched blocks");
     let best_peers: Vec<_> = peers.get_best_proved_peers(best_tip);
     let last_hash = best_tip.calc_header_hash();
 
@@ -33,9 +34,10 @@ pub(crate) fn prove_or_download_matched_blocks(
                 peers.get_matched_blocks_to_prove(matched_blocks, GET_BLOCKS_PROOF_LIMIT);
             if !blocks_to_prove.is_empty() {
                 debug!(
-                    "send get blocks proof request to peer: {}, count={}",
+                    "send get blocks proof request to peer: {}, count={}, {:?}",
                     peer_index,
-                    blocks_to_prove.len()
+                    blocks_to_prove.len(),
+                    blocks_to_prove,
                 );
                 let content = packed::GetBlocksProof::new_builder()
                     .block_hashes(blocks_to_prove.pack())