async fn check_state_of_current_utxos(&mut self) -> Result<(), HorizonSyncError> {
        let config = self.shared.config.horizon_sync_config;
        let local_tip_height = self.local_metadata.height_of_longest_chain();
        let local_num_utxo_nodes = async_db::fetch_mmr_node_count(self.db(), MmrTree::Utxo, local_tip_height).await?;

        debug!(
            target: LOG_TARGET,
            "Checking current utxo state between {} and {}", 0, local_num_utxo_nodes
        );

        let chunks = self.chunked_count_iter(0, local_num_utxo_nodes, config.max_utxo_mmr_node_request_size);
        for (pos, count) in chunks {
            let num_sync_peers = self.sync_peers.len();
            for attempt in 1..=num_sync_peers {
                let (remote_utxo_hashes, remote_utxo_deleted, sync_peer) = helpers::request_mmr_nodes(
                    LOG_TARGET,
                    self.shared,
                    self.sync_peers,
                    MmrTree::Utxo,
                    pos,
                    count,
                    self.horizon_sync_height,
                    config.max_sync_request_retry_attempts,
                )
                .await?;
                let (local_utxo_hashes, local_utxo_bitmap_bytes) = self
                    .shared
                    .local_node_interface
                    .fetch_mmr_nodes(MmrTree::Utxo, pos, count, self.horizon_sync_height)
                    .await?;
                let local_utxo_deleted = Bitmap::deserialize(&local_utxo_bitmap_bytes);

                match self.validate_utxo_hashes_response(&remote_utxo_hashes, &local_utxo_hashes) {
                    Ok(_) => {
                        let num_hashes = local_utxo_hashes.len();
                        let spent_utxos = local_utxo_hashes
                            .into_iter()
                            .enumerate()
                            .filter_map(|(index, hash)| {
                                let deleted_index = pos + index as u32;
                                let local_deleted = local_utxo_deleted.contains(deleted_index);
                                let remote_deleted = remote_utxo_deleted.contains(deleted_index);
                                if remote_deleted && !local_deleted {
                                    Some(hash)
                                } else {
                                    None
                                }
                            })
                            .collect::<Vec<_>>();

                        let num_deleted = spent_utxos.len();
                        async_db::horizon_sync_spend_utxos(self.db(), spent_utxos).await?;

                        debug!(
                            target: LOG_TARGET,
                            "Checked {} existing UTXO(s). Marked {} UTXO(s) as spent.", num_hashes, num_deleted
                        );

                        break;
                    },
                    Err(err @ HorizonSyncError::IncorrectResponse) => {
                        warn!(
                            target: LOG_TARGET,
                            "Invalid UTXO hashes received from peer `{}`: {}", sync_peer, err
                        );
                        // Exclude the peer (without banning) as they could be on the wrong chain
                        exclude_sync_peer(LOG_TARGET, self.sync_peers, &sync_peer)?;
                    },
                    Err(e) => return Err(e),
                };
                debug!(target: LOG_TARGET, "Retrying UTXO state check. Attempt {}", attempt);
                if attempt == num_sync_peers {
                    return Err(HorizonSyncError::MaxSyncAttemptsReached);
                }
            }
        }

        Ok(())
    }
// Checks if any existing UTXOs in the local database have been spent according to the remote state
Checks if any existing UTXOs in the local database have been spent according to the remote state
Checks if any existing UTXOs in the local database have been spent according to the remote state
async fn check_state_of_current_utxos(&mut self) -> Result<(), HorizonSyncError> {
        let config = self.shared.config.horizon_sync_config;
        let local_tip_height = self.local_metadata.height_of_longest_chain();
        let local_num_utxo_nodes = async_db::fetch_mmr_node_count(self.db(), MmrTree::Utxo, local_tip_height).await?;
        debug!(
            target: LOG_TARGET,
            "Checking current utxo state between {} and {}", 0, local_num_utxo_nodes
        );
        let chunks = self.chunked_count_iter(0, local_num_utxo_nodes, config.max_utxo_mmr_node_request_size);
        for (pos, count) in chunks {
            let num_sync_peers = self.sync_peers.len();
            for attempt in 1..=num_sync_peers {
                let (remote_utxo_hashes, remote_utxo_deleted, sync_peer) = helpers::request_mmr_nodes(
                    LOG_TARGET,
                    self.shared,
                    self.sync_peers,
                    MmrTree::Utxo,
                    pos,
                    count,
                    self.horizon_sync_height,
                    config.max_sync_request_retry_attempts,
                )
                .await?;
                let (local_utxo_hashes, local_utxo_bitmap_bytes) = self
                    .shared
                    .local_node_interface
                    .fetch_mmr_nodes(MmrTree::Utxo, pos, count, self.horizon_sync_height)
                    .await?;
                let local_utxo_deleted = Bitmap::deserialize(&local_utxo_bitmap_bytes);
                match self.validate_utxo_hashes_response(&remote_utxo_hashes, &local_utxo_hashes) {
                    Ok(_) => {
                        let num_hashes = local_utxo_hashes.len();
                        let spent_utxos = local_utxo_hashes
                            .into_iter()
                            .enumerate()
                            .filter_map(|(index, hash)| {
                                let deleted_index = pos + index as u32;
                                let local_deleted = local_utxo_deleted.contains(deleted_index);
                                let remote_deleted = remote_utxo_deleted.contains(deleted_index);
                                if remote_deleted && !local_deleted {
                                    Some(hash)
                                } else {
                                    None
                                }
                            })
                            .collect::<Vec<_>>();
                        let num_deleted = spent_utxos.len();
                        async_db::horizon_sync_spend_utxos(self.db(), spent_utxos).await?;
                        debug!(
                            target: LOG_TARGET,
                            "Checked {} existing UTXO(s). Marked {} UTXO(s) as spent.", num_hashes, num_deleted
                        );
                        break;
                    },
                    Err(err @ HorizonSyncError::IncorrectResponse) => {
                        warn!(
                            target: LOG_TARGET,
                            "Invalid UTXO hashes received from peer `{}`: {}", sync_peer, err
                        );
                        // Exclude the peer (without banning) as they could be on the wrong chain
                        exclude_sync_peer(LOG_TARGET, self.sync_peers, &sync_peer)?;
                    },
                    Err(e) => return Err(e),
                };
                debug!(target: LOG_TARGET, "Retrying UTXO state check. Attempt {}", attempt);
                if attempt == num_sync_peers {
                    return Err(HorizonSyncError::MaxSyncAttemptsReached);
                }
            }
        }
        Ok(())
    }
async fn check_state_of_current_utxos ( & mut self ) -> Result < ( ) , HorizonSyncError > { let config = self . shared . config . horizon_sync_config ; let local_tip_height = self . local_metadata . height_of_longest_chain ( ) ; let local_num_utxo_nodes = async_db :: fetch_mmr_node_count ( self . db ( ) , MmrTree :: Utxo , local_tip_height ) . await ? ; debug ! ( target LOG_TARGET " " 0 local_num_utxo_nodes ) ; let chunks = self . chunked_count_iter ( 0 , local_num_utxo_nodes , config . max_utxo_mmr_node_request_size ) ; for ( pos , count ) in chunks { let num_sync_peers = self . sync_peers . len ( ) ; for attempt in 1 ..= num_sync_peers { let ( remote_utxo_hashes , remote_utxo_deleted , sync_peer ) = helpers :: request_mmr_nodes ( LOG_TARGET , self . shared , self . sync_peers , MmrTree :: Utxo , pos , count , self . horizon_sync_height , config . max_sync_request_retry_attempts , ) . await ? ; let ( local_utxo_hashes , local_utxo_bitmap_bytes ) = self . shared . local_node_interface . fetch_mmr_nodes ( MmrTree :: Utxo , pos , count , self . horizon_sync_height ) . await ? ; let local_utxo_deleted = Bitmap :: deserialize ( & local_utxo_bitmap_bytes ) ; match self . validate_utxo_hashes_response ( & remote_utxo_hashes , & local_utxo_hashes ) { Ok ( _ ) => { let num_hashes = local_utxo_hashes . len ( ) ; let spent_utxos = local_utxo_hashes . into_iter ( ) . enumerate ( ) . filter_map ( | ( index , hash ) | { let deleted_index = pos + index as u32 ; let local_deleted = local_utxo_deleted . contains ( deleted_index ) ; let remote_deleted = remote_utxo_deleted . contains ( deleted_index ) ; if remote_deleted && ! local_deleted { Some ( hash ) } else { None } } ) . collect :: < Vec < _ > > ( ) ; let num_deleted = spent_utxos . len ( ) ; async_db :: horizon_sync_spend_utxos ( self . db ( ) , spent_utxos ) . await ? ; debug ! ( target LOG_TARGET " " num_hashes num_deleted ) ; break ; } , Err ( err @ HorizonSyncError :: IncorrectResponse ) => { warn ! ( target LOG_TARGET " " sync_peer err ) ; // Exclude the peer (without banning) as they could be on the wrong chain exclude_sync_peer ( LOG_TARGET , self . sync_peers , & sync_peer ) ? ; } , Err ( e ) => return Err ( e ) , } ; debug ! ( target LOG_TARGET " " attempt ) ; if attempt == num_sync_peers { return Err ( HorizonSyncError :: MaxSyncAttemptsReached ) ; } } } Ok ( ( ) ) }
Checks if any existing UTXOs in the local database have been spent according to the remote state
Checks if any existing UTXOs in the local database have been spent according to the remote state
&mut self
&mut self