Merge pull request #257 from eval-exec/exec/fix-clean-up-matched-blocks-at-startup

 fix: Handle missing blocks to prevent sync stall

Author: quake

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

@@ -143,22 +143,76 @@ impl FilterProtocol {
             debug!("found best proved peer {}", peer);
 
             let mut matched_blocks = self.peers.matched_blocks().write().await;
-            if let Some((db_start_number, blocks_count, db_blocks)) =
-                self.storage.get_earliest_matched_blocks()
-            {
+            if let Some(db_matched_blocks) = self.storage.get_earliest_matched_blocks() {
                 debug!(
                     "try recover matched blocks from storage, start_number={}, \
                              blocks_count={}, matched_count: {}",
-                    db_start_number,
-                    blocks_count,
+                    db_matched_blocks.start_number,
+                    db_matched_blocks.blocks_count,
                     matched_blocks.len(),
                 );
                 let option = matched_blocks.is_empty();
 
                 if option {
-                    // recover matched blocks from storage
+                    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, db_blocks);
+                        .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),
@@ -174,6 +228,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!(
@@ -262,10 +321,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();
@@ -226,12 +228,15 @@ impl<'a> BlockFiltersProcess<'a> {
             );
             let option = matched_blocks.is_empty();
             if option {
-                if let Some((_start_number, _blocks_count, db_blocks)) =
-                    self.filter.storage.get_earliest_matched_blocks()
-                {
-                    self.filter
-                        .peers
-                        .add_matched_blocks(&mut matched_blocks, db_blocks);
+                if let Some(db_matched_blocks) = self.filter.storage.get_earliest_matched_blocks() {
+                    self.filter.peers.add_matched_blocks(
+                        &mut matched_blocks,
+                        db_matched_blocks
+                            .blocks
+                            .into_iter()
+                            .map(|b| (b.hash, b.proved))
+                            .collect(),
+                    );
                     prove_or_download_matched_blocks(
                         Arc::clone(&self.filter.peers),
                         &tip_header,

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/mod.rs

@@ -355,10 +355,10 @@ impl LightClientProtocol {
                     info!("rollback to block#1 since previous last header number is 1");
                     let mut matched_blocks = self.peers.matched_blocks().write().await;
 
-                    while let Some((start_number, _, _)) = self.storage.get_latest_matched_blocks()
-                    {
-                        if start_number > 0 {
-                            self.storage.remove_matched_blocks(start_number);
+                    while let Some(matched_blocks_data) = self.storage.get_latest_matched_blocks() {
+                        if matched_blocks_data.start_number > 0 {
+                            self.storage
+                                .remove_matched_blocks(matched_blocks_data.start_number);
                         }
                     }
                     self.storage.rollback_to_block(1);
@@ -385,18 +385,19 @@ impl LightClientProtocol {
                     debug!("fork to number: {}", to_number);
                     let mut matched_blocks = self.peers.matched_blocks().write().await;
                     let mut start_number_opt = None;
-                    while let Some((start_number, blocks_count, _)) =
-                        self.storage.get_latest_matched_blocks()
-                    {
+                    while let Some(matched_blocks_data) = self.storage.get_latest_matched_blocks() {
                         // Remove matched blocks if the range contains blocks after the fork point
                         // The range is [start_number, start_number + blocks_count - 1]
                         // Fork point (to_number) is the last valid block, so remove ranges containing blocks > to_number
-                        if start_number + blocks_count > to_number + 1 {
+                        if matched_blocks_data.start_number + matched_blocks_data.blocks_count
+                            > to_number + 1
+                        {
                             debug!("remove matched blocks start from: {} (range covers {} blocks, contains blocks after fork at {})",
-                                   start_number, blocks_count, to_number);
-                            self.storage.remove_matched_blocks(start_number);
+                                   matched_blocks_data.start_number, matched_blocks_data.blocks_count, to_number);
+                            self.storage
+                                .remove_matched_blocks(matched_blocks_data.start_number);
                         } else {
-                            start_number_opt = Some(start_number);
+                            start_number_opt = Some(matched_blocks_data.start_number);
                             break;
                         }
                     }
@@ -704,6 +705,11 @@ impl LightClientProtocol {
     }
 
     async fn get_idle_blocks(&mut self, nc: &BoxedCKBProtocolContext) {
+        // Clean up old missing headers (uncle blocks) older than 1 hour
+        // to prevent unbounded memory growth in fetching_headers
+        const MAX_MISSING_AGE_MS: u64 = 3_600_000; // 1 hour
+        self.peers.cleanup_old_missing_headers(MAX_MISSING_AGE_MS);
+
         let tip_header = self.storage.get_tip_header();
         let matched_blocks = self.peers.matched_blocks().read().await;
 

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]) {
@@ -1228,6 +1242,32 @@ impl Peers {
             }
         }
     }
+
+    /// Clean up old missing entries from fetching_headers to prevent unbounded memory growth
+    /// This is called periodically to remove uncle blocks that were marked as missing
+    /// but are no longer needed (older than max_age_ms)
+    pub(crate) fn cleanup_old_missing_headers(&self, max_age_ms: u64) {
+        let now = unix_time_as_millis();
+        let mut removed_count = 0;
+
+        self.fetching_headers.retain(|_hash, info| {
+            if info.missing && now.saturating_sub(info.added_ts) > max_age_ms {
+                removed_count += 1;
+                false // Remove this entry
+            } else {
+                true // Keep this entry
+            }
+        });
+
+        if removed_count > 0 {
+            log::debug!(
+                "Cleaned up {} old missing headers (age > {}ms)",
+                removed_count,
+                max_age_ms
+            );
+        }
+    }
+
     // mark all fetching hashes (headers/txs) as timeout
     pub(crate) fn mark_fetching_headers_timeout(&self, peer_index: PeerIndex) {
         if let Some(peer) = self.get_peer(&peer_index) {

light-client-lib/src/protocols/synchronizer.rs

@@ -65,12 +65,17 @@ impl CKBProtocolHandler for SyncProtocol {
                 if !matched_blocks.is_empty()
                     && self.peers.all_matched_blocks_downloaded(&matched_blocks)
                 {
-                    let (start_number, blocks_count, db_blocks) = self
+                    let matched_blocks_data = self
                         .storage
                         .get_earliest_matched_blocks()
                         .expect("get matched blocks from storage");
-                    let db_blocks: HashSet<_> =
-                        db_blocks.into_iter().map(|(hash, _)| hash).collect();
+                    let start_number = matched_blocks_data.start_number;
+                    let blocks_count = matched_blocks_data.blocks_count;
+                    let db_blocks: HashSet<_> = matched_blocks_data
+                        .blocks
+                        .into_iter()
+                        .map(|b| b.hash)
+                        .collect();
 
                     self.storage.remove_matched_blocks(start_number);
                     let blocks = self.peers.clear_matched_blocks(&mut matched_blocks);
@@ -90,11 +95,15 @@ impl CKBProtocolHandler for SyncProtocol {
                         .update_block_number(start_number + blocks_count - 1);
 
                     // send more GetBlocksProof/GetBlocks requests
-                    if let Some((_start_number, _blocks_count, db_blocks)) =
-                        self.storage.get_earliest_matched_blocks()
-                    {
-                        self.peers
-                            .add_matched_blocks(&mut matched_blocks, db_blocks);
+                    if let Some(db_matched_blocks) = self.storage.get_earliest_matched_blocks() {
+                        self.peers.add_matched_blocks(
+                            &mut matched_blocks,
+                            db_matched_blocks
+                                .blocks
+                                .into_iter()
+                                .map(|b| (b.hash, b.proved))
+                                .collect(),
+                        );
                         let tip_header = self.storage.get_tip_header();
                         prove_or_download_matched_blocks(
                             Arc::clone(&self.peers),