diff --git a/client/network/src/protocol.rs b/client/network/src/protocol.rs index c1887ce35bfdb..ac74af0f5ca94 100644 --- a/client/network/src/protocol.rs +++ b/client/network/src/protocol.rs @@ -122,6 +122,8 @@ mod rep { pub const BAD_ROLE: Rep = Rep::new_fatal("Unsupported role"); /// Peer response data does not have requested bits. pub const BAD_RESPONSE: Rep = Rep::new(-(1 << 12), "Incomplete response"); + /// Peer send us a block announcement that failed at validation. + pub const BAD_BLOCK_ANNOUNCEMENT: Rep = Rep::new(-(1 << 12), "Bad block announcement"); } struct Metrics { @@ -542,7 +544,9 @@ impl Protocol { pub fn update_chain(&mut self) { let info = self.context_data.chain.info(); self.sync.update_chain_info(&info.best_hash, info.best_number); - self.behaviour.set_legacy_handshake_message(build_status_message(&self.config, &self.context_data.chain)); + self.behaviour.set_legacy_handshake_message( + build_status_message(&self.config, &self.context_data.chain), + ); self.behaviour.set_notif_protocol_handshake( &self.block_announces_protocol, BlockAnnouncesHandshake::build(&self.config, &self.context_data.chain).encode() @@ -568,16 +572,21 @@ impl Protocol { self.context_data.peers.iter().map(|(id, peer)| (id, &peer.info)) } - pub fn on_custom_message( + fn on_custom_message( &mut self, who: PeerId, data: BytesMut, ) -> CustomMessageOutcome { - let message = match as Decode>::decode(&mut &data[..]) { Ok(message) => message, Err(err) => { - debug!(target: "sync", "Couldn't decode packet sent by {}: {:?}: {}", who, data, err.what()); + debug!( + target: "sync", + "Couldn't decode packet sent by {}: {:?}: {}", + who, + data, + err.what(), + ); self.peerset_handle.report_peer(who, rep::BAD_MESSAGE); return CustomMessageOutcome::None; } @@ -590,11 +599,8 @@ impl Protocol { match message { GenericMessage::Status(_) => debug!(target: "sub-libp2p", "Received unexpected Status"), - GenericMessage::BlockAnnounce(announce) => { - let outcome = self.on_block_announce(who.clone(), announce); - self.update_peer_info(&who); - return outcome; - }, + GenericMessage::BlockAnnounce(announce) => + self.push_block_announce_validation(who.clone(), announce), GenericMessage::Transactions(m) => self.on_transactions(who, m), GenericMessage::BlockResponse(_) => @@ -1156,42 +1162,73 @@ impl Protocol { } } - fn on_block_announce( + /// Push a block announce validation. + /// + /// It is required that [`ChainSync::poll_block_announce_validation`] is + /// called later to check for finished validations. The result of the validation + /// needs to be passed to [`Protocol::process_block_announce_validation_result`] + /// to finish the processing. + /// + /// # Note + /// + /// This will internally create a future, but this future will not be registered + /// in the task before being polled once. So, it is required to call + /// [`ChainSync::poll_block_announce_validation`] to ensure that the future is + /// registered properly and will wake up the task when being ready. + fn push_block_announce_validation( &mut self, who: PeerId, announce: BlockAnnounce, - ) -> CustomMessageOutcome { + ) { let hash = announce.header.hash(); - let number = *announce.header.number(); if let Some(ref mut peer) = self.context_data.peers.get_mut(&who) { peer.known_blocks.insert(hash.clone()); } - let is_their_best = match announce.state.unwrap_or(message::BlockState::Best) { + let is_best = match announce.state.unwrap_or(message::BlockState::Best) { message::BlockState::Best => true, message::BlockState::Normal => false, }; - match self.sync.on_block_announce(&who, hash, &announce, is_their_best) { - sync::OnBlockAnnounce::Nothing => { + self.sync.push_block_announce_validation(who, hash, announce, is_best); + } + + /// Process the result of the block announce validation. + fn process_block_announce_validation_result( + &mut self, + validation_result: sync::PollBlockAnnounceValidation, + ) -> CustomMessageOutcome { + let (header, is_best, who) = match validation_result { + sync::PollBlockAnnounceValidation::Nothing { is_best, who, header } => { + self.update_peer_info(&who); + // `on_block_announce` returns `OnBlockAnnounce::ImportHeader` // when we have all data required to import the block // in the BlockAnnounce message. This is only when: // 1) we're on light client; // AND // 2) parent block is already imported and not pruned. - if is_their_best { - return CustomMessageOutcome::PeerNewBest(who, number); + if is_best { + return CustomMessageOutcome::PeerNewBest(who, *header.number()) } else { - return CustomMessageOutcome::None; + return CustomMessageOutcome::None } } - sync::OnBlockAnnounce::ImportHeader => () // We proceed with the import. - } + sync::PollBlockAnnounceValidation::ImportHeader { header, is_best, who } => { + self.update_peer_info(&who); + (header, is_best, who) + } + sync::PollBlockAnnounceValidation::Failure { who } => { + self.report_peer(who, rep::BAD_BLOCK_ANNOUNCEMENT); + return CustomMessageOutcome::None + } + }; + + let number = *header.number(); - // to import header from announced block let's construct response to request that normally would have - // been sent over network (but it is not in our case) + // to import header from announced block let's construct response to request that normally + // would have been sent over network (but it is not in our case) let blocks_to_import = self.sync.on_block_data( &who, None, @@ -1199,8 +1236,8 @@ impl Protocol { id: 0, blocks: vec![ message::generic::BlockData { - hash: hash, - header: Some(announce.header), + hash: header.hash(), + header: Some(header), body: None, receipt: None, message_queue: None, @@ -1210,8 +1247,10 @@ impl Protocol { }, ); - if is_their_best { - self.pending_messages.push_back(CustomMessageOutcome::PeerNewBest(who, number)); + if is_best { + self.pending_messages.push_back( + CustomMessageOutcome::PeerNewBest(who, number), + ); } match blocks_to_import { @@ -1549,6 +1588,15 @@ impl NetworkBehaviour for Protocol { warn!(target: "sub-libp2p", "Inconsistent state, no peers for pending transaction!"); } } + + // Check if there is any block announcement validation finished. + while let Poll::Ready(result) = self.sync.poll_block_announce_validation(cx) { + match self.process_block_announce_validation_result(result) { + CustomMessageOutcome::None => {}, + outcome => self.pending_messages.push_back(outcome), + } + } + if let Some(message) = self.pending_messages.pop_front() { return Poll::Ready(NetworkBehaviourAction::GenerateEvent(message)); } @@ -1643,9 +1691,15 @@ impl NetworkBehaviour for Protocol { } Some(Fallback::BlockAnnounce) => { if let Ok(announce) = message::BlockAnnounce::decode(&mut message.as_ref()) { - let outcome = self.on_block_announce(peer_id.clone(), announce); - self.update_peer_info(&peer_id); - outcome + self.push_block_announce_validation(peer_id, announce); + + // Make sure that the newly added block announce validation future was + // polled once to be registered in the task. + if let Poll::Ready(res) = self.sync.poll_block_announce_validation(cx) { + self.process_block_announce_validation_result(res) + } else { + CustomMessageOutcome::None + } } else { warn!(target: "sub-libp2p", "Failed to decode block announce"); CustomMessageOutcome::None diff --git a/client/network/src/protocol/sync.rs b/client/network/src/protocol/sync.rs index bfd8c4fe218de..9866dc0568a8e 100644 --- a/client/network/src/protocol/sync.rs +++ b/client/network/src/protocol/sync.rs @@ -49,7 +49,11 @@ use sp_runtime::{ traits::{Block as BlockT, Header, NumberFor, Zero, One, CheckedSub, SaturatedConversion, Hash, HashFor} }; use sp_arithmetic::traits::Saturating; -use std::{fmt, ops::Range, collections::{HashMap, HashSet, VecDeque}, sync::Arc}; +use std::{ + fmt, ops::Range, collections::{HashMap, hash_map::Entry, HashSet, VecDeque}, + sync::Arc, pin::Pin, +}; +use futures::{task::Poll, Future, stream::FuturesUnordered, FutureExt, StreamExt}; mod blocks; mod extra_requests; @@ -63,6 +67,17 @@ const MAX_IMPORTING_BLOCKS: usize = 2048; /// Maximum blocks to download ahead of any gap. const MAX_DOWNLOAD_AHEAD: u32 = 2048; +/// Maximum number of concurrent block announce validations. +/// +/// If the queue reaches the maximum, we drop any new block +/// announcements. +const MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS: usize = 256; + +/// Maximum number of concurrent block announce validations per peer. +/// +/// See [`MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS`] for more information. +const MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS_PER_PEER: usize = 4; + /// We use a heuristic that with a high likelihood, by the time /// `MAJOR_SYNC_BLOCKS` have been imported we'll be on the same /// chain as (or at least closer to) the peer so we want to delay @@ -192,6 +207,12 @@ pub struct ChainSync { max_parallel_downloads: u32, /// Total number of downloaded blocks. downloaded_blocks: usize, + /// All block announcement that are currently being validated. + block_announce_validation: FuturesUnordered< + Pin> + Send>> + >, + /// Stats per peer about the number of concurrent block announce validations. + block_announce_validation_per_peer_stats: HashMap, } /// All the data we have about a Peer that we are trying to sync with @@ -306,13 +327,65 @@ pub enum OnBlockData { Request(PeerId, BlockRequest) } -/// Result of [`ChainSync::on_block_announce`]. +/// Result of [`ChainSync::poll_block_announce_validation`]. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum OnBlockAnnounce { +pub enum PollBlockAnnounceValidation { + /// The announcement failed at validation. + /// + /// The peer reputation should be decreased. + Failure { + /// Who sent the processed block announcement? + who: PeerId, + }, /// The announcement does not require further handling. - Nothing, + Nothing { + /// Who sent the processed block announcement? + who: PeerId, + /// Was this their new best block? + is_best: bool, + /// The header of the announcement. + header: H, + }, /// The announcement header should be imported. - ImportHeader, + ImportHeader { + /// Who sent the processed block announcement? + who: PeerId, + /// Was this their new best block? + is_best: bool, + /// The header of the announcement. + header: H, + }, +} + +/// Result of [`ChainSync::block_announce_validation`]. +#[derive(Debug, Clone, PartialEq, Eq)] +enum PreValidateBlockAnnounce { + /// The announcement failed at validation. + /// + /// The peer reputation should be decreased. + Failure { + /// Who sent the processed block announcement? + who: PeerId, + }, + /// The announcement does not require further handling. + Nothing { + /// Who sent the processed block announcement? + who: PeerId, + /// Was this their new best block? + is_best: bool, + /// The announcement. + announce: BlockAnnounce, + }, + /// The pre-validation was sucessful and the announcement should be + /// further processed. + Process { + /// Is this the new best block of the peer? + is_new_best: bool, + /// The id of the peer that send us the announcement. + who: PeerId, + /// The announcement. + announce: BlockAnnounce, + }, } /// Result of [`ChainSync::on_block_justification`]. @@ -343,6 +416,16 @@ pub enum OnBlockFinalityProof { } } +/// Result of [`ChainSync::has_slot_for_block_announce_validation`]. +enum HasSlotForBlockAnnounceValidation { + /// Yes, there is a slot for the block announce validation. + Yes, + /// We reached the total maximum number of validation slots. + TotalMaximumSlotsReached, + /// We reached the maximum number of validation slots for the given peer. + MaximumPeerSlotsReached, +} + impl ChainSync { /// Create a new instance. pub fn new( @@ -377,6 +460,8 @@ impl ChainSync { block_announce_validator, max_parallel_downloads, downloaded_blocks: 0, + block_announce_validation: Default::default(), + block_announce_validation_per_peer_stats: Default::default(), } } @@ -1156,60 +1241,215 @@ impl ChainSync { self.pending_requests.set_all(); } - /// Call when a node announces a new block. + /// Checks if there is a slot for a block announce validation. /// - /// If `OnBlockAnnounce::ImportHeader` is returned, then the caller MUST try to import passed - /// header (call `on_block_data`). The network request isn't sent - /// in this case. Both hash and header is passed as an optimization - /// to avoid rehashing the header. - pub fn on_block_announce(&mut self, who: &PeerId, hash: B::Hash, announce: &BlockAnnounce, is_best: bool) - -> OnBlockAnnounce - { + /// The total number and the number per peer of concurrent block announce validations + /// is capped. + /// + /// Returns [`HasSlotForBlockAnnounceValidation`] to inform about the result. + fn has_slot_for_block_announce_validation(&mut self, peer: &PeerId) -> HasSlotForBlockAnnounceValidation { + if self.block_announce_validation.len() >= MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS { + return HasSlotForBlockAnnounceValidation::TotalMaximumSlotsReached + } + + match self.block_announce_validation_per_peer_stats.entry(peer.clone()) { + Entry::Vacant(entry) => { + entry.insert(1); + HasSlotForBlockAnnounceValidation::Yes + }, + Entry::Occupied(mut entry) => { + if *entry.get() < MAX_CONCURRENT_BLOCK_ANNOUNCE_VALIDATIONS_PER_PEER { + *entry.get_mut() += 1; + HasSlotForBlockAnnounceValidation::Yes + } else { + HasSlotForBlockAnnounceValidation::MaximumPeerSlotsReached + } + }, + } + } + + /// Push a block announce validation. + /// + /// It is required that [`ChainSync::poll_block_announce_validation`] is called + /// to check for finished block announce validations. + pub fn push_block_announce_validation( + &mut self, + who: PeerId, + hash: B::Hash, + announce: BlockAnnounce, + is_best: bool, + ) { let header = &announce.header; let number = *header.number(); - debug!(target: "sync", "Received block announcement {:?} with number {:?} from {}", hash, number, who); + debug!( + target: "sync", + "Pre-validating received block announcement {:?} with number {:?} from {}", + hash, + number, + who, + ); + if number.is_zero() { - warn!(target: "sync", "💔 Ignored genesis block (#0) announcement from {}: {}", who, hash); - return OnBlockAnnounce::Nothing + self.block_announce_validation.push(async move { + warn!( + target: "sync", + "💔 Ignored genesis block (#0) announcement from {}: {}", + who, + hash, + ); + PreValidateBlockAnnounce::Nothing { is_best, who, announce } + }.boxed()); + return + } + + // Check if there is a slot for this block announce validation. + match self.has_slot_for_block_announce_validation(&who) { + HasSlotForBlockAnnounceValidation::Yes => {}, + HasSlotForBlockAnnounceValidation::TotalMaximumSlotsReached => { + self.block_announce_validation.push(async move { + warn!( + target: "sync", + "💔 Ignored block (#{} -- {}) announcement from {} because all validation slots are occupied.", + number, + hash, + who, + ); + PreValidateBlockAnnounce::Nothing { is_best, who, announce } + }.boxed()); + return + } + HasSlotForBlockAnnounceValidation::MaximumPeerSlotsReached => { + self.block_announce_validation.push(async move { + warn!( + target: "sync", + "💔 Ignored block (#{} -- {}) announcement from {} because all validation slots for this peer are occupied.", + number, + hash, + who, + ); + PreValidateBlockAnnounce::Nothing { is_best, who, announce } + }.boxed()); + return + } } - let parent_status = self.block_status(header.parent_hash()).ok().unwrap_or(BlockStatus::Unknown); + + // Let external validator check the block announcement. + let assoc_data = announce.data.as_ref().map_or(&[][..], |v| v.as_slice()); + let future = self.block_announce_validator.validate(&header, assoc_data); + let hash = hash.clone(); + + self.block_announce_validation.push(async move { + match future.await { + Ok(Validation::Success { is_new_best }) => PreValidateBlockAnnounce::Process { + is_new_best: is_new_best || is_best, + announce, + who, + }, + Ok(Validation::Failure) => { + debug!( + target: "sync", + "Block announcement validation of block {} from {} failed", + hash, + who, + ); + PreValidateBlockAnnounce::Failure { who } + } + Err(e) => { + error!(target: "sync", "💔 Block announcement validation errored: {}", e); + PreValidateBlockAnnounce::Nothing { is_best, who, announce } + } + } + }.boxed()); + } + + /// Poll block announce validation. + /// + /// Block announce validations can be pushed by using + /// [`ChainSync::push_block_announce_validation`]. + /// + /// This should be polled until it returns [`Poll::Pending`]. + /// + /// If [`PollBlockAnnounceValidation::ImportHeader`] is returned, then the caller MUST try to import passed + /// header (call `on_block_data`). The network request isn't sent in this case. + pub fn poll_block_announce_validation( + &mut self, + cx: &mut std::task::Context, + ) -> Poll> { + match self.block_announce_validation.poll_next_unpin(cx) { + Poll::Ready(Some(res)) => Poll::Ready(self.finish_block_announce_validation(res)), + _ => Poll::Pending, + } + } + + /// Should be called when a block announce validation was finished, to update the stats + /// of the given peer. + fn peer_block_announce_validation_finished(&mut self, peer: &PeerId) { + match self.block_announce_validation_per_peer_stats.entry(peer.clone()) { + Entry::Vacant(_) => { + error!( + target: "sync", + "💔 Block announcement validation from peer {} finished for that no slot was allocated!", + peer, + ); + }, + Entry::Occupied(mut entry) => { + if entry.get_mut().saturating_sub(1) == 0 { + entry.remove(); + } + } + } + } + + /// This will finish processing of the block announcement. + fn finish_block_announce_validation( + &mut self, + pre_validation_result: PreValidateBlockAnnounce, + ) -> PollBlockAnnounceValidation { + let (announce, is_best, who) = match pre_validation_result { + PreValidateBlockAnnounce::Nothing { is_best, who, announce } => { + self.peer_block_announce_validation_finished(&who); + return PollBlockAnnounceValidation::Nothing { is_best, who, header: announce.header } + }, + PreValidateBlockAnnounce::Failure { who } => { + self.peer_block_announce_validation_finished(&who); + return PollBlockAnnounceValidation::Failure { who } + }, + PreValidateBlockAnnounce::Process { announce, is_new_best, who } => { + self.peer_block_announce_validation_finished(&who); + (announce, is_new_best, who) + }, + }; + + let header = announce.header; + let number = *header.number(); + let hash = header.hash(); + let parent_status = self.block_status(header.parent_hash()).unwrap_or(BlockStatus::Unknown); let known_parent = parent_status != BlockStatus::Unknown; let ancient_parent = parent_status == BlockStatus::InChainPruned; let known = self.is_known(&hash); - let peer = if let Some(peer) = self.peers.get_mut(who) { + let peer = if let Some(peer) = self.peers.get_mut(&who) { peer } else { error!(target: "sync", "💔 Called on_block_announce with a bad peer ID"); - return OnBlockAnnounce::Nothing + return PollBlockAnnounceValidation::Nothing { is_best, who, header } }; + while peer.recently_announced.len() >= ANNOUNCE_HISTORY_SIZE { peer.recently_announced.pop_front(); } peer.recently_announced.push_back(hash.clone()); - // Let external validator check the block announcement. - let assoc_data = announce.data.as_ref().map_or(&[][..], |v| v.as_slice()); - let is_best = match self.block_announce_validator.validate(&header, assoc_data) { - Ok(Validation::Success { is_new_best }) => is_new_best || is_best, - Ok(Validation::Failure) => { - debug!(target: "sync", "Block announcement validation of block {} from {} failed", hash, who); - return OnBlockAnnounce::Nothing - } - Err(e) => { - error!(target: "sync", "💔 Block announcement validation errored: {}", e); - return OnBlockAnnounce::Nothing - } - }; - if is_best { // update their best block peer.best_number = number; peer.best_hash = hash; } + if let PeerSyncState::AncestorSearch {..} = peer.state { - return OnBlockAnnounce::Nothing + return PollBlockAnnounceValidation::Nothing { is_best, who, header } } + // If the announced block is the best they have and is not ahead of us, our common number // is either one further ahead or it's the one they just announced, if we know about it. if is_best { @@ -1221,7 +1461,7 @@ impl ChainSync { peer.common_number = number - One::one(); } } - self.pending_requests.add(who); + self.pending_requests.add(&who); // known block case if known || self.is_already_downloading(&hash) { @@ -1229,18 +1469,18 @@ impl ChainSync { if let Some(target) = self.fork_targets.get_mut(&hash) { target.peers.insert(who.clone()); } - return OnBlockAnnounce::Nothing + return PollBlockAnnounceValidation::Nothing { is_best, who, header } } if ancient_parent { trace!(target: "sync", "Ignored ancient block announced from {}: {} {:?}", who, hash, header); - return OnBlockAnnounce::Nothing + return PollBlockAnnounceValidation::Nothing { is_best, who, header } } let requires_additional_data = !self.role.is_light() || !known_parent; if !requires_additional_data { trace!(target: "sync", "Importing new header announced from {}: {} {:?}", who, hash, header); - return OnBlockAnnounce::ImportHeader + return PollBlockAnnounceValidation::ImportHeader { is_best, header, who } } if number <= self.best_queued_number { @@ -1258,7 +1498,7 @@ impl ChainSync { .peers.insert(who.clone()); } - OnBlockAnnounce::Nothing + PollBlockAnnounceValidation::Nothing { is_best, who, header } } /// Call when a peer has disconnected. diff --git a/client/network/test/src/sync.rs b/client/network/test/src/sync.rs index 86e274aae10eb..64985871d85e0 100644 --- a/client/network/test/src/sync.rs +++ b/client/network/test/src/sync.rs @@ -18,7 +18,7 @@ use sp_consensus::BlockOrigin; use std::time::Duration; -use futures::executor::block_on; +use futures::{Future, executor::block_on}; use super::*; use sp_consensus::block_validation::Validation; use substrate_test_runtime::Header; @@ -693,14 +693,7 @@ fn can_sync_to_peers_with_wrong_common_block() { let fork_hash = net.peer(0).push_blocks_at(BlockId::Number(0), 2, false); net.peer(1).push_blocks_at(BlockId::Number(0), 2, false); // wait for connection - block_on(futures::future::poll_fn::<(), _>(|cx| { - net.poll(cx); - if net.peer(0).num_peers() == 0 || net.peer(1).num_peers() == 0 { - Poll::Pending - } else { - Poll::Ready(()) - } - })); + net.block_until_connected(); // both peers re-org to the same fork without notifying each other net.peer(0).client().finalize_block(BlockId::Hash(fork_hash), Some(Vec::new()), true).unwrap(); @@ -720,8 +713,8 @@ impl BlockAnnounceValidator for NewBestBlockAnnounceValidator { &mut self, _: &Header, _: &[u8], - ) -> Result> { - Ok(Validation::Success { is_new_best: true }) + ) -> Pin>> + Send>> { + async { Ok(Validation::Success { is_new_best: true }) }.boxed() } } @@ -750,3 +743,39 @@ fn sync_blocks_when_block_announce_validator_says_it_is_new_best() { // that flags all blocks as `is_new_best` and thus, it should have synced the blocks. assert!(!net.peer(1).has_block(&block_hash)); } + +/// Waits for some time until the validation is successfull. +struct DeferredBlockAnnounceValidator; + +impl BlockAnnounceValidator for DeferredBlockAnnounceValidator { + fn validate( + &mut self, + _: &Header, + _: &[u8], + ) -> Pin>> + Send>> { + async { + futures_timer::Delay::new(std::time::Duration::from_millis(500)).await; + Ok(Validation::Success { is_new_best: false }) + }.boxed() + } +} + +#[test] +fn wait_until_deferred_block_announce_validation_is_ready() { + sp_tracing::try_init_simple(); + log::trace!(target: "sync", "Test"); + let mut net = TestNet::with_fork_choice(ForkChoiceStrategy::Custom(false)); + net.add_full_peer_with_config(Default::default()); + net.add_full_peer_with_config(FullPeerConfig { + block_announce_validator: Some(Box::new(NewBestBlockAnnounceValidator)), + ..Default::default() + }); + + net.block_until_connected(); + + let block_hash = net.peer(0).push_blocks(1, true); + + while !net.peer(1).has_block(&block_hash) { + net.block_until_idle(); + } +} diff --git a/primitives/consensus/common/src/block_validation.rs b/primitives/consensus/common/src/block_validation.rs index 66f960f16fff3..f8255130e6416 100644 --- a/primitives/consensus/common/src/block_validation.rs +++ b/primitives/consensus/common/src/block_validation.rs @@ -18,7 +18,8 @@ use crate::BlockStatus; use sp_runtime::{generic::BlockId, traits::Block}; -use std::{error::Error, sync::Arc}; +use std::{error::Error, future::Future, pin::Pin, sync::Arc}; +use futures::FutureExt as _; /// A type which provides access to chain information. pub trait Chain { @@ -47,7 +48,16 @@ pub enum Validation { /// Type which checks incoming block announcements. pub trait BlockAnnounceValidator { /// Validate the announced header and its associated data. - fn validate(&mut self, header: &B::Header, data: &[u8]) -> Result>; + /// + /// # Note + /// + /// Returning [`Validation::Failure`] will lead to a decrease of the + /// peers reputation as it sent us invalid data. + fn validate( + &mut self, + header: &B::Header, + data: &[u8], + ) -> Pin>> + Send>>; } /// Default implementation of `BlockAnnounceValidator`. @@ -55,7 +65,11 @@ pub trait BlockAnnounceValidator { pub struct DefaultBlockAnnounceValidator; impl BlockAnnounceValidator for DefaultBlockAnnounceValidator { - fn validate(&mut self, _h: &B::Header, _d: &[u8]) -> Result> { - Ok(Validation::Success { is_new_best: false }) + fn validate( + &mut self, + _: &B::Header, + _: &[u8], + ) -> Pin>> + Send>> { + async { Ok(Validation::Success { is_new_best: false }) }.boxed() } }