// Synchronize kernels upto the horizon sync height from remote sync peers.
Synchronize kernels upto the horizon sync height from remote sync peers.
Synchronize kernels upto the horizon sync height from remote sync peers .
async fn synchronize_kernels(&mut self) -> Result<(), HorizonSyncError> {
        let config = self.shared.config.horizon_sync_config;
        let local_num_kernels =
            async_db::fetch_mmr_node_count(self.db(), MmrTree::Kernel, self.horizon_sync_height).await?;
        let (remote_num_kernels, sync_peer) = helpers::request_mmr_node_count(
            LOG_TARGET,
            self.shared,
            self.sync_peers,
            MmrTree::Kernel,
            self.horizon_sync_height,
            config.max_sync_request_retry_attempts,
        )
        .await?;
        if local_num_kernels >= remote_num_kernels {
            debug!(target: LOG_TARGET, "Local kernel set already synchronized");
            return Ok(());
        }
        debug!(
            target: LOG_TARGET,
            "Requesting kernels from {} to {} ({} remaining)",
            local_num_kernels,
            remote_num_kernels,
            remote_num_kernels - local_num_kernels,
        );
        let chunks = self.chunked_count_iter(
            local_num_kernels,
            remote_num_kernels,
            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 (kernel_hashes, _, sync_peer1) = helpers::request_mmr_nodes(
                    LOG_TARGET,
                    self.shared,
                    self.sync_peers,
                    MmrTree::Kernel,
                    pos,
                    count,
                    self.horizon_sync_height,
                    config.max_sync_request_retry_attempts,
                )
                .await?;
                let (kernels, sync_peer2) = helpers::request_kernels(
                    LOG_TARGET,
                    self.shared,
                    self.sync_peers,
                    kernel_hashes.clone(),
                    config.max_sync_request_retry_attempts,
                )
                .await?;
                match self.validate_kernel_response(&kernel_hashes, &kernels) {
                    Ok(_) => {
                        let num_kernels = kernels.len();
                        async_db::horizon_sync_insert_kernels(self.db(), kernels).await?;
                        trace!(
                            target: LOG_TARGET,
                            "{} kernels successfully added to database ({} remaining)",
                            num_kernels,
                            remote_num_kernels - pos,
                        );
                        break;
                    },
                    Err(err @ HorizonSyncError::EmptyResponse { .. }) |
                    Err(err @ HorizonSyncError::IncorrectResponse { .. }) |
                    Err(err @ HorizonSyncError::InvalidKernelSignature(_)) => {
                        warn!(target: LOG_TARGET, "{}", err);
                        // TODO: Fetching mmr nodes and kernels should both be attempted for the same peer
                        if sync_peer1 == sync_peer2 {
                            debug!(
                                target: LOG_TARGET,
                                "Banning peer {} from local node, because they supplied invalid kernels", sync_peer
                            );
                            self.ban_sync_peer(&sync_peer, "Peer supplied invalid kernels".to_string())
                                .await?;
                        }
                    },
                    Err(e) => return Err(e),
                };
                debug!(
                    target: LOG_TARGET,
                    "Retrying kernel sync. Attempt {} of {}", attempt, config.max_sync_request_retry_attempts
                );
                if attempt == num_sync_peers {
                    return Err(HorizonSyncError::MaxSyncAttemptsReached);
                }
            }
        }
        self.validate_mmr_root(MmrTree::Kernel).await?;
        Ok(())
    }
async fn synchronize_kernels ( & mut self ) -> Result < ( ) , HorizonSyncError > { let config = self . shared . config . horizon_sync_config ; let local_num_kernels = async_db :: fetch_mmr_node_count ( self . db ( ) , MmrTree :: Kernel , self . horizon_sync_height ) . await ? ; let ( remote_num_kernels , sync_peer ) = helpers :: request_mmr_node_count ( LOG_TARGET , self . shared , self . sync_peers , MmrTree :: Kernel , self . horizon_sync_height , config . max_sync_request_retry_attempts , ) . await ? ; if local_num_kernels >= remote_num_kernels { debug ! ( target LOG_TARGET " " ) ; return Ok ( ( ) ) ; } debug ! ( target LOG_TARGET " " local_num_kernels remote_num_kernels remote_num_kernels local_num_kernels ) ; let chunks = self . chunked_count_iter ( local_num_kernels , remote_num_kernels , 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 ( kernel_hashes , _ , sync_peer1 ) = helpers :: request_mmr_nodes ( LOG_TARGET , self . shared , self . sync_peers , MmrTree :: Kernel , pos , count , self . horizon_sync_height , config . max_sync_request_retry_attempts , ) . await ? ; let ( kernels , sync_peer2 ) = helpers :: request_kernels ( LOG_TARGET , self . shared , self . sync_peers , kernel_hashes . clone ( ) , config . max_sync_request_retry_attempts , ) . await ? ; match self . validate_kernel_response ( & kernel_hashes , & kernels ) { Ok ( _ ) => { let num_kernels = kernels . len ( ) ; async_db :: horizon_sync_insert_kernels ( self . db ( ) , kernels ) . await ? ; trace ! ( target LOG_TARGET " " num_kernels remote_num_kernels pos ) ; break ; } , Err ( err @ HorizonSyncError :: EmptyResponse { .. } ) | Err ( err @ HorizonSyncError :: IncorrectResponse { .. } ) | Err ( err @ HorizonSyncError :: InvalidKernelSignature ( _ ) ) => { warn ! ( target LOG_TARGET " " err ) ; // TODO: Fetching mmr nodes and kernels should both be attempted for the same peer if sync_peer1 == sync_peer2 { debug ! ( target LOG_TARGET " " sync_peer ) ; self . ban_sync_peer ( & sync_peer , " " . to_string ( ) ) . await ? ; } } , Err ( e ) => return Err ( e ) , } ; debug ! ( target LOG_TARGET " " attempt config max_sync_request_retry_attempts ) ; if attempt == num_sync_peers { return Err ( HorizonSyncError :: MaxSyncAttemptsReached ) ; } } } self . validate_mmr_root ( MmrTree :: Kernel ) . await ? ; Ok ( ( ) ) }
Synchronize kernels upto the horizon sync height from remote sync peers.
Synchronize kernels upto the horizon sync height from remote sync peers .
&mut self