async fn request_blocks<B: BlockchainBackend + 'static>(
    shared: &mut BaseNodeStateMachine<B>,
    sync_peers: &mut SyncPeers,
    block_nums: &[u64],
) -> Result<(Vec<Arc<Block>>, SyncPeer), BlockSyncError>
{
    let config = shared.config.sync_peer_config;
    for attempt in 1..=shared.config.block_sync_config.max_block_request_retry_attempts {
        let sync_peer = select_sync_peer(&config, sync_peers)?;
        if block_nums.is_empty() {
            return Ok((Vec::new(), sync_peer));
        }
        debug!(
            target: LOG_TARGET,
            "Requesting blocks {:?} from {}.", block_nums, sync_peer
        );
        if let StateInfo::BlockSync(ref mut info) = shared.info {
            info.local_height = block_nums[0];
            info.tip_height = block_nums[block_nums.len() - 1];
        }
        shared.publish_event_info();
        match shared
            .outbound_nci
            .request_blocks_from_peer(block_nums.to_vec(), Some(sync_peer.node_id.clone()))
            .await
        {
            Ok(hist_blocks) => {
                debug!(target: LOG_TARGET, "Received {} blocks from peer", hist_blocks.len());
                if block_nums.len() == hist_blocks.len() {
                    if (0..block_nums.len()).all(|i| hist_blocks[i].block().header.height == block_nums[i]) {
                        let blocks = hist_blocks
                            .into_iter()
                            .map(|hist_block| Arc::new(hist_block.into_block()))
                            .collect::<Vec<_>>();
                        return Ok((blocks, sync_peer));
                    }

                    debug!(
                        target: LOG_TARGET,
                        "Banning peer {} from local node, because they supplied the incorrect blocks", sync_peer
                    );
                    ban_sync_peer(
                        LOG_TARGET,
                        &mut shared.connectivity,
                        sync_peers,
                        &sync_peer,
                        config.short_term_peer_ban_duration,
                        "Peer supplied the incorrect blocks".to_string(),
                    )
                    .await?;
                } else {
                    debug!(
                        target: LOG_TARGET,
                        "Incorrect number of blocks returned. Expected {}. Got {}",
                        block_nums.len(),
                        hist_blocks.len()
                    );
                    debug!(
                        target: LOG_TARGET,
                        "Banning peer {} from local node, because they supplied the incorrect number of blocks",
                        sync_peer
                    );
                    ban_sync_peer(
                        LOG_TARGET,
                        &mut shared.connectivity,
                        sync_peers,
                        &sync_peer,
                        config.short_term_peer_ban_duration,
                        "Peer supplied the incorrect number of blocks".to_string(),
                    )
                    .await?;
                }
            },
            Err(CommsInterfaceError::UnexpectedApiResponse) => {
                debug!(target: LOG_TARGET, "Remote node provided an unexpected api response.",);
                ban_sync_peer(
                    LOG_TARGET,
                    &mut shared.connectivity,
                    sync_peers,
                    &sync_peer,
                    config.peer_ban_duration,
                    "Remote node provided an unexpected api response".to_string(),
                )
                .await?;
            },
            Err(CommsInterfaceError::RequestTimedOut) => {
                debug!(
                    target: LOG_TARGET,
                    "Failed to fetch blocks from peer: {:?}. Retrying.",
                    CommsInterfaceError::RequestTimedOut,
                );
                ban_sync_peer(
                    LOG_TARGET,
                    &mut shared.connectivity,
                    sync_peers,
                    &sync_peer,
                    config.short_term_peer_ban_duration,
                    "Failed to fetch blocks from peer".to_string(),
                )
                .await?;
            },
            Err(e) => return Err(BlockSyncError::CommsInterfaceError(e)),
        }
        debug!(target: LOG_TARGET, "Retrying block download. Attempt {}", attempt);
    }
    Err(BlockSyncError::MaxRequestAttemptsReached)
}
// Request a block from a remote sync peer.
Request a block from a remote sync peer.
Request a block from a remote sync peer .
async fn request_blocks<B: BlockchainBackend + 'static>(
    shared: &mut BaseNodeStateMachine<B>,
    sync_peers: &mut SyncPeers,
    block_nums: &[u64],
) -> Result<(Vec<Arc<Block>>, SyncPeer), BlockSyncError>
{
    let config = shared.config.sync_peer_config;
    for attempt in 1..=shared.config.block_sync_config.max_block_request_retry_attempts {
        let sync_peer = select_sync_peer(&config, sync_peers)?;
        if block_nums.is_empty() {
            return Ok((Vec::new(), sync_peer));
        }
        debug!(
            target: LOG_TARGET,
            "Requesting blocks {:?} from {}.", block_nums, sync_peer
        );
        if let StateInfo::BlockSync(ref mut info) = shared.info {
            info.local_height = block_nums[0];
            info.tip_height = block_nums[block_nums.len() - 1];
        }
        shared.publish_event_info();
        match shared
            .outbound_nci
            .request_blocks_from_peer(block_nums.to_vec(), Some(sync_peer.node_id.clone()))
            .await
        {
            Ok(hist_blocks) => {
                debug!(target: LOG_TARGET, "Received {} blocks from peer", hist_blocks.len());
                if block_nums.len() == hist_blocks.len() {
                    if (0..block_nums.len()).all(|i| hist_blocks[i].block().header.height == block_nums[i]) {
                        let blocks = hist_blocks
                            .into_iter()
                            .map(|hist_block| Arc::new(hist_block.into_block()))
                            .collect::<Vec<_>>();
                        return Ok((blocks, sync_peer));
                    }
                    debug!(
                        target: LOG_TARGET,
                        "Banning peer {} from local node, because they supplied the incorrect blocks", sync_peer
                    );
                    ban_sync_peer(
                        LOG_TARGET,
                        &mut shared.connectivity,
                        sync_peers,
                        &sync_peer,
                        config.short_term_peer_ban_duration,
                        "Peer supplied the incorrect blocks".to_string(),
                    )
                    .await?;
                } else {
                    debug!(
                        target: LOG_TARGET,
                        "Incorrect number of blocks returned. Expected {}. Got {}",
                        block_nums.len(),
                        hist_blocks.len()
                    );
                    debug!(
                        target: LOG_TARGET,
                        "Banning peer {} from local node, because they supplied the incorrect number of blocks",
                        sync_peer
                    );
                    ban_sync_peer(
                        LOG_TARGET,
                        &mut shared.connectivity,
                        sync_peers,
                        &sync_peer,
                        config.short_term_peer_ban_duration,
                        "Peer supplied the incorrect number of blocks".to_string(),
                    )
                    .await?;
                }
            },
            Err(CommsInterfaceError::UnexpectedApiResponse) => {
                debug!(target: LOG_TARGET, "Remote node provided an unexpected api response.",);
                ban_sync_peer(
                    LOG_TARGET,
                    &mut shared.connectivity,
                    sync_peers,
                    &sync_peer,
                    config.peer_ban_duration,
                    "Remote node provided an unexpected api response".to_string(),
                )
                .await?;
            },
            Err(CommsInterfaceError::RequestTimedOut) => {
                debug!(
                    target: LOG_TARGET,
                    "Failed to fetch blocks from peer: {:?}. Retrying.",
                    CommsInterfaceError::RequestTimedOut,
                );
                ban_sync_peer(
                    LOG_TARGET,
                    &mut shared.connectivity,
                    sync_peers,
                    &sync_peer,
                    config.short_term_peer_ban_duration,
                    "Failed to fetch blocks from peer".to_string(),
                )
                .await?;
            },
            Err(e) => return Err(BlockSyncError::CommsInterfaceError(e)),
        }
        debug!(target: LOG_TARGET, "Retrying block download. Attempt {}", attempt);
    }
    Err(BlockSyncError::MaxRequestAttemptsReached)
}
async fn request_blocks < B : BlockchainBackend + ' static > ( shared : & mut BaseNodeStateMachine < B > , sync_peers : & mut SyncPeers , block_nums : & [ u64 ] , ) -> Result < ( Vec < Arc < Block > > , SyncPeer ) , BlockSyncError > { let config = shared . config . sync_peer_config ; for attempt in 1 ..= shared . config . block_sync_config . max_block_request_retry_attempts { let sync_peer = select_sync_peer ( & config , sync_peers ) ? ; if block_nums . is_empty ( ) { return Ok ( ( Vec :: new ( ) , sync_peer ) ) ; } debug ! ( target LOG_TARGET " " block_nums sync_peer ) ; if let StateInfo :: BlockSync ( ref mut info ) = shared . info { info . local_height = block_nums [ 0 ] ; info . tip_height = block_nums [ block_nums . len ( ) - 1 ] ; } shared . publish_event_info ( ) ; match shared . outbound_nci . request_blocks_from_peer ( block_nums . to_vec ( ) , Some ( sync_peer . node_id . clone ( ) ) ) . await { Ok ( hist_blocks ) => { debug ! ( target LOG_TARGET " " hist_blocks len ( ) ) ; if block_nums . len ( ) == hist_blocks . len ( ) { if ( 0 .. block_nums . len ( ) ) . all ( | i | hist_blocks [ i ] . block ( ) . header . height == block_nums [ i ] ) { let blocks = hist_blocks . into_iter ( ) . map ( | hist_block | Arc :: new ( hist_block . into_block ( ) ) ) . collect :: < Vec < _ > > ( ) ; return Ok ( ( blocks , sync_peer ) ) ; } debug ! ( target LOG_TARGET " " sync_peer ) ; ban_sync_peer ( LOG_TARGET , & mut shared . connectivity , sync_peers , & sync_peer , config . short_term_peer_ban_duration , " " . to_string ( ) , ) . await ? ; } else { debug ! ( target LOG_TARGET " " block_nums len ( ) hist_blocks len ( ) ) ; debug ! ( target LOG_TARGET " " sync_peer ) ; ban_sync_peer ( LOG_TARGET , & mut shared . connectivity , sync_peers , & sync_peer , config . short_term_peer_ban_duration , " " . to_string ( ) , ) . await ? ; } } , Err ( CommsInterfaceError :: UnexpectedApiResponse ) => { debug ! ( target LOG_TARGET " " ) ; ban_sync_peer ( LOG_TARGET , & mut shared . connectivity , sync_peers , & sync_peer , config . peer_ban_duration , " " . to_string ( ) , ) . await ? ; } , Err ( CommsInterfaceError :: RequestTimedOut ) => { debug ! ( target LOG_TARGET " " CommsInterfaceError RequestTimedOut ) ; ban_sync_peer ( LOG_TARGET , & mut shared . connectivity , sync_peers , & sync_peer , config . short_term_peer_ban_duration , " " . to_string ( ) , ) . await ? ; } , Err ( e ) => return Err ( BlockSyncError :: CommsInterfaceError ( e ) ) , } debug ! ( target LOG_TARGET " " attempt ) ; } Err ( BlockSyncError :: MaxRequestAttemptsReached ) }
Request a block from a remote sync peer.
Request a block from a remote sync peer .
shared
&mut BaseNodeStateMachine<B>
sync_peers
&mut SyncPeers
block_nums
&[u64]
shared
&mut BaseNodeStateMachine<B>
sync_peers
&mut SyncPeers
block_nums
&[u64]