// Synchronize UTXO MMR Nodes, RangeProof MMR Nodes and the UTXO set upto the horizon sync height from
// remote sync peers.
Synchronize UTXO MMR Nodes, RangeProof MMR Nodes and the UTXO set upto the horizon sync height from
remote sync peers.
Synchronize UTXO MMR Nodes RangeProof MMR Nodes and the UTXO set upto the horizon sync height from remote sync peers .
async fn synchronize_utxos_and_rangeproofs(&mut self) -> Result<(), HorizonSyncError> {
        let config = self.shared.config.horizon_sync_config;
        let local_num_utxo_nodes =
            async_db::fetch_mmr_node_count(self.shared.db.clone(), MmrTree::Utxo, self.horizon_sync_height).await?;
        let (remote_num_utxo_nodes, _sync_peer) = helpers::request_mmr_node_count(
            LOG_TARGET,
            self.shared,
            self.sync_peers,
            MmrTree::Utxo,
            self.horizon_sync_height,
            config.max_sync_request_retry_attempts,
        )
        .await?;
        if local_num_utxo_nodes >= remote_num_utxo_nodes {
            debug!(target: LOG_TARGET, "UTXOs and range proofs are already synchronized.");
            return Ok(());
        }
        debug!(
            target: LOG_TARGET,
            "Synchronizing {} UTXO MMR nodes from {} to {}",
            remote_num_utxo_nodes - local_num_utxo_nodes,
            local_num_utxo_nodes,
            remote_num_utxo_nodes
        );
        let chunks = self.chunked_count_iter(
            local_num_utxo_nodes,
            remote_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 (utxo_hashes, utxo_bitmap, sync_peer1) = 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 (rp_hashes, _, sync_peer2) = helpers::request_mmr_nodes(
                    LOG_TARGET,
                    self.shared,
                    self.sync_peers,
                    MmrTree::RangeProof,
                    pos,
                    count,
                    self.horizon_sync_height,
                    config.max_sync_request_retry_attempts,
                )
                .await?;
                // Construct the list of hashes of the UTXOs that need to be requested.
                let mut request_utxo_hashes = Vec::new();
                let mut request_rp_hashes = Vec::new();
                let mut is_stxos = Vec::with_capacity(utxo_hashes.len());
                for index in 0..utxo_hashes.len() {
                    let deleted = utxo_bitmap.contains(pos + index as u32);
                    is_stxos.push(deleted);
                    if !deleted {
                        request_utxo_hashes.push(&utxo_hashes[index]);
                        request_rp_hashes.push(&rp_hashes[index]);
                    }
                }
                // Download a partial UTXO set
                let (mut utxos, sync_peer3) = helpers::request_txos(
                    LOG_TARGET,
                    self.shared,
                    self.sync_peers,
                    &request_utxo_hashes,
                    config.max_sync_request_retry_attempts,
                )
                .await?;
                debug!(
                    target: LOG_TARGET,
                    "Fetched {} UTXOs ({} were not downloaded because they are spent)",
                    utxos.len(),
                    is_stxos.iter().filter(|x| **x).count()
                );
                let db = &self.shared.db;
                match self.validate_utxo_and_rangeproof_response(
                    &utxo_hashes,
                    &rp_hashes,
                    &request_utxo_hashes,
                    &request_rp_hashes,
                    &utxos,
                ) {
                    Ok(_) => {
                        // The order of these inserts are important to ensure the MMRs are constructed correctly
                        // and the roots match.
                        for (index, is_stxo) in is_stxos.into_iter().enumerate() {
                            if is_stxo {
                                async_db::insert_mmr_node(db.clone(), MmrTree::Utxo, utxo_hashes[index].clone(), true)
                                    .await?;
                                async_db::insert_mmr_node(
                                    db.clone(),
                                    MmrTree::RangeProof,
                                    rp_hashes[index].clone(),
                                    false,
                                )
                                .await?;
                            } else {
                                // Inserting the UTXO will also insert the corresponding UTXO and RangeProof MMR
                                // Nodes.
                                async_db::insert_utxo(db.clone(), utxos.remove(0)).await?;
                            }
                        }
                        async_db::horizon_sync_create_mmr_checkpoint(self.db(), MmrTree::Utxo).await?;
                        async_db::horizon_sync_create_mmr_checkpoint(self.db(), MmrTree::RangeProof).await?;
                        trace!(
                            target: LOG_TARGET,
                            "{} UTXOs with MMR nodes inserted into database",
                            utxo_hashes.len()
                        );
                        break;
                    },
                    Err(err @ HorizonSyncError::EmptyResponse { .. }) |
                    Err(err @ HorizonSyncError::IncorrectResponse { .. }) => {
                        warn!(
                            target: LOG_TARGET,
                            "Invalid UTXOs or MMR Nodes received from peer. {}", err
                        );
                        if (sync_peer1 == sync_peer2) && (sync_peer1 == sync_peer3) {
                            debug!(
                                target: LOG_TARGET,
                                "Banning peer {} from local node, because they supplied invalid UTXOs or MMR Nodes",
                                sync_peer1
                            );
                            self.ban_sync_peer(&sync_peer1, "Peer supplied invalid UTXOs or MMR Nodes".to_string())
                                .await?;
                        }
                    },
                    Err(e) => return Err(e),
                };
                debug!(target: LOG_TARGET, "Retrying kernel sync. Attempt {}", attempt);
                if attempt == num_sync_peers {
                    return Err(HorizonSyncError::MaxSyncAttemptsReached);
                }
            }
        }
        self.validate_mmr_root(MmrTree::Utxo).await?;
        self.validate_mmr_root(MmrTree::RangeProof).await?;
        Ok(())
    }
async fn synchronize_utxos_and_rangeproofs ( & mut self ) -> Result < ( ) , HorizonSyncError > { let config = self . shared . config . horizon_sync_config ; let local_num_utxo_nodes = async_db :: fetch_mmr_node_count ( self . shared . db . clone ( ) , MmrTree :: Utxo , self . horizon_sync_height ) . await ? ; let ( remote_num_utxo_nodes , _sync_peer ) = helpers :: request_mmr_node_count ( LOG_TARGET , self . shared , self . sync_peers , MmrTree :: Utxo , self . horizon_sync_height , config . max_sync_request_retry_attempts , ) . await ? ; if local_num_utxo_nodes >= remote_num_utxo_nodes { debug ! ( target LOG_TARGET " " ) ; return Ok ( ( ) ) ; } debug ! ( target LOG_TARGET " " remote_num_utxo_nodes local_num_utxo_nodes local_num_utxo_nodes remote_num_utxo_nodes ) ; let chunks = self . chunked_count_iter ( local_num_utxo_nodes , remote_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 ( utxo_hashes , utxo_bitmap , sync_peer1 ) = 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 ( rp_hashes , _ , sync_peer2 ) = helpers :: request_mmr_nodes ( LOG_TARGET , self . shared , self . sync_peers , MmrTree :: RangeProof , pos , count , self . horizon_sync_height , config . max_sync_request_retry_attempts , ) . await ? ; // Construct the list of hashes of the UTXOs that need to be requested. let mut request_utxo_hashes = Vec :: new ( ) ; let mut request_rp_hashes = Vec :: new ( ) ; let mut is_stxos = Vec :: with_capacity ( utxo_hashes . len ( ) ) ; for index in 0 .. utxo_hashes . len ( ) { let deleted = utxo_bitmap . contains ( pos + index as u32 ) ; is_stxos . push ( deleted ) ; if ! deleted { request_utxo_hashes . push ( & utxo_hashes [ index ] ) ; request_rp_hashes . push ( & rp_hashes [ index ] ) ; } } // Download a partial UTXO set let ( mut utxos , sync_peer3 ) = helpers :: request_txos ( LOG_TARGET , self . shared , self . sync_peers , & request_utxo_hashes , config . max_sync_request_retry_attempts , ) . await ? ; debug ! ( target LOG_TARGET " " utxos len ( ) is_stxos iter ( ) filter ( x x ) count ( ) ) ; let db = & self . shared . db ; match self . validate_utxo_and_rangeproof_response ( & utxo_hashes , & rp_hashes , & request_utxo_hashes , & request_rp_hashes , & utxos , ) { Ok ( _ ) => { // The order of these inserts are important to ensure the MMRs are constructed correctly // and the roots match. for ( index , is_stxo ) in is_stxos . into_iter ( ) . enumerate ( ) { if is_stxo { async_db :: insert_mmr_node ( db . clone ( ) , MmrTree :: Utxo , utxo_hashes [ index ] . clone ( ) , true ) . await ? ; async_db :: insert_mmr_node ( db . clone ( ) , MmrTree :: RangeProof , rp_hashes [ index ] . clone ( ) , false , ) . await ? ; } else { // Inserting the UTXO will also insert the corresponding UTXO and RangeProof MMR // Nodes. async_db :: insert_utxo ( db . clone ( ) , utxos . remove ( 0 ) ) . await ? ; } } async_db :: horizon_sync_create_mmr_checkpoint ( self . db ( ) , MmrTree :: Utxo ) . await ? ; async_db :: horizon_sync_create_mmr_checkpoint ( self . db ( ) , MmrTree :: RangeProof ) . await ? ; trace ! ( target LOG_TARGET " " utxo_hashes len ( ) ) ; break ; } , Err ( err @ HorizonSyncError :: EmptyResponse { .. } ) | Err ( err @ HorizonSyncError :: IncorrectResponse { .. } ) => { warn ! ( target LOG_TARGET " " err ) ; if ( sync_peer1 == sync_peer2 ) && ( sync_peer1 == sync_peer3 ) { debug ! ( target LOG_TARGET " " sync_peer1 ) ; self . ban_sync_peer ( & sync_peer1 , " " . to_string ( ) ) . await ? ; } } , Err ( e ) => return Err ( e ) , } ; debug ! ( target LOG_TARGET " " attempt ) ; if attempt == num_sync_peers { return Err ( HorizonSyncError :: MaxSyncAttemptsReached ) ; } } } self . validate_mmr_root ( MmrTree :: Utxo ) . await ? ; self . validate_mmr_root ( MmrTree :: RangeProof ) . await ? ; Ok ( ( ) ) }
Synchronize UTXO MMR Nodes, RangeProof MMR Nodes and the UTXO set upto the horizon sync height from
remote sync peers.
Synchronize UTXO MMR Nodes RangeProof MMR Nodes and the UTXO set upto the horizon sync height from remote sync peers .
&mut self