diff --git a/core/network/src/protocol.rs b/core/network/src/protocol.rs index b2cea2cd5c82b..ef581a6e43c3d 100644 --- a/core/network/src/protocol.rs +++ b/core/network/src/protocol.rs @@ -1236,6 +1236,13 @@ impl, H: ExHashT> Protocol { self.sync.request_justification(&hash, number) } + /// Request syncing for the given block from given set of peers. + /// Uses `protocol` to queue a new block download request and tries to dispatch all pending + /// requests. + pub fn set_sync_fork_request(&mut self, peers: Vec, hash: &B::Hash, number: NumberFor) { + self.sync.set_sync_fork_request(peers, hash, number) + } + /// A batch of blocks have been processed, with or without errors. /// Call this when a batch of blocks have been processed by the importqueue, with or without /// errors. diff --git a/core/network/src/protocol/sync.rs b/core/network/src/protocol/sync.rs index 6a2a207abcdf8..39927792beab8 100644 --- a/core/network/src/protocol/sync.rs +++ b/core/network/src/protocol/sync.rs @@ -123,7 +123,10 @@ pub struct ChainSync { queue_blocks: HashSet, /// The best block number that we are currently importing. best_importing_number: NumberFor, + /// Finality proof handler. request_builder: Option>, + /// Explicit sync requests. + sync_requests: HashMap>, /// A flag that caches idle state with no pending requests. is_idle: bool, /// A type to check incoming block announcements. @@ -157,6 +160,11 @@ pub struct PeerInfo { pub best_number: NumberFor } +struct SyncRequest { + number: NumberFor, + peers: HashSet, +} + /// The state of syncing between a Peer and ourselves. /// /// Generally two categories, "busy" or `Available`. If busy, the enum @@ -299,6 +307,7 @@ impl ChainSync { queue_blocks: Default::default(), best_importing_number: Zero::zero(), request_builder, + sync_requests: Default::default(), is_idle: false, block_announce_validator, } @@ -449,6 +458,51 @@ impl ChainSync { }) } + /// Request syncing for the given block from given set of peers. + // The implementation is similar to on_block_announce with unknown parent hash. + pub fn set_sync_fork_request(&mut self, peers: Vec, hash: &B::Hash, number: NumberFor) { + if peers.is_empty() { + if let Some(_) = self.sync_requests.remove(hash) { + debug!(target: "sync", "Cleared sync request for block {:?} with {:?}", hash, peers); + } + return; + } + debug!(target: "sync", "Explicit sync request for block {:?} with {:?}", hash, peers); + if self.is_known(&hash) { + debug!(target: "sync", "Refusing to sync known hash {:?}", hash); + return; + } + + let block_status = self.client.block_status(&BlockId::Number(number - One::one())) + .unwrap_or(BlockStatus::Unknown); + if block_status == BlockStatus::InChainPruned { + trace!(target: "sync", "Refusing to sync ancient block {:?}", hash); + return; + } + + self.is_idle = false; + for peer_id in &peers { + if let Some(peer) = self.peers.get_mut(peer_id) { + if let PeerSyncState::AncestorSearch(_, _) = peer.state { + continue; + } + + if number > peer.best_number { + peer.best_number = number; + peer.best_hash = hash.clone(); + } + } + } + + self.sync_requests + .entry(hash.clone()) + .or_insert_with(|| SyncRequest { + number, + peers: Default::default(), + }) + .peers.extend(peers); + } + /// Get an iterator over all scheduled justification requests. pub fn justification_requests(&mut self) -> impl Iterator)> + '_ { let peers = &mut self.peers; @@ -508,13 +562,21 @@ impl ChainSync { } let blocks = &mut self.blocks; let attrs = &self.required_block_attributes; + let sync_requests = &self.sync_requests; let mut have_requests = false; + let last_finalized = self.client.info().chain.finalized_number; + let best_queued = self.best_queued_number; let iter = self.peers.iter_mut().filter_map(move |(id, peer)| { if !peer.state.is_available() { trace!(target: "sync", "Peer {} is busy", id); return None } - if let Some((range, req)) = peer_block_request(id, peer, blocks, attrs) { + if let Some((hash, req)) = explicit_sync_request(id, sync_requests, best_queued, last_finalized, attrs) { + trace!(target: "sync", "Downloading explicitly requested block {:?} from {}", hash, id); + peer.state = PeerSyncState::DownloadingStale(hash); + have_requests = true; + Some((id.clone(), req)) + } else if let Some((range, req)) = peer_block_request(id, peer, blocks, attrs) { peer.state = PeerSyncState::DownloadingNew(range.start); trace!(target: "sync", "New block request for {}", id); have_requests = true; @@ -860,6 +922,9 @@ impl ChainSync { self.best_queued_number = number; self.best_queued_hash = *hash; } + if let Some(_) = self.sync_requests.remove(&hash) { + trace!(target: "sync", "Completed explicit sync request {:?}", hash); + } // Update common blocks for (n, peer) in self.peers.iter_mut() { if let PeerSyncState::AncestorSearch(_, _) = peer.state { @@ -1232,3 +1297,32 @@ fn peer_block_request( None } } + +/// Get pending explicit sync request for a peer. +fn explicit_sync_request( + id: &PeerId, + requests: &HashMap>, + best_num: NumberFor, + finalized: NumberFor, + attributes: &message::BlockAttributes, +) -> Option<(B::Hash, BlockRequest)> +{ + for (hash, r) in requests { + if !r.peers.contains(id) { + continue + } + if r.number <= best_num { + trace!(target: "sync", "Downloading requested fork {:?} from {}", hash, id); + return Some((hash.clone(), message::generic::BlockRequest { + id: 0, + fields: attributes.clone(), + from: message::FromBlock::Hash(hash.clone()), + to: None, + direction: message::Direction::Descending, + max: Some((r.number - finalized).saturated_into::()), // up to the last finalized block + })) + } + } + None +} + diff --git a/core/network/src/service.rs b/core/network/src/service.rs index 3ca7bffdb45c5..2cf949116f012 100644 --- a/core/network/src/service.rs +++ b/core/network/src/service.rs @@ -496,6 +496,18 @@ impl, H: ExHashT> NetworkServic Ok(()) } + /// Configure an explicit fork sync request. + /// Note that this function should not be used for recent blocks. + /// Sync should be able to download all the recent forks normally. + /// `set_sync_fork_request` should only be used if external code detects that there's + /// a stale fork missing. + /// Passing empty `peers` set effectively removes the sync request. + pub fn set_sync_fork_request(&self, peers: Vec, hash: B::Hash, number: NumberFor) { + let _ = self + .to_worker + .unbounded_send(ServerToWorkerMsg::SyncFork(peers, hash, number)); + } + /// Modify a peerset priority group. pub fn set_priority_group(&self, group_id: String, peers: HashSet) -> Result<(), String> { let peers = peers.into_iter().map(|p| { @@ -586,6 +598,7 @@ enum ServerToWorkerMsg> { GetValue(record::Key), PutValue(record::Key, Vec), AddKnownAddress(PeerId, Multiaddr), + SyncFork(Vec, B::Hash, NumberFor), } /// Main network worker. Must be polled in order for the network to advance. @@ -664,6 +677,8 @@ impl, H: ExHashT> Stream for Ne self.network_service.put_value(key, value), ServerToWorkerMsg::AddKnownAddress(peer_id, addr) => self.network_service.add_known_address(peer_id, addr), + ServerToWorkerMsg::SyncFork(peer_ids, hash, number) => + self.network_service.user_protocol_mut().set_sync_fork_request(peer_ids, &hash, number), } } diff --git a/core/network/src/test/mod.rs b/core/network/src/test/mod.rs index ddc86b6e95e18..302ef78d7d43d 100644 --- a/core/network/src/test/mod.rs +++ b/core/network/src/test/mod.rs @@ -229,6 +229,11 @@ pub struct Peer> { } impl> Peer { + /// Get this peer ID. + pub fn id(&self) -> PeerId { + self.network.service().local_peer_id() + } + /// Returns true if we're major syncing. pub fn is_major_syncing(&self) -> bool { self.network.service().is_major_syncing() @@ -259,6 +264,11 @@ impl> Peer { self.network.service().announce_block(hash, data); } + /// Request explicit fork sync. + pub fn set_sync_fork_request(&self, peers: Vec, hash: ::Hash, number: NumberFor) { + self.network.service().set_sync_fork_request(peers, hash, number); + } + /// Add blocks to the peer -- edit the block before adding pub fn generate_blocks(&mut self, count: usize, origin: BlockOrigin, edit_block: F) -> H256 where F: FnMut(BlockBuilder) -> Block diff --git a/core/network/src/test/sync.rs b/core/network/src/test/sync.rs index d50190f6573ff..3d8e57cad038b 100644 --- a/core/network/src/test/sync.rs +++ b/core/network/src/test/sync.rs @@ -526,3 +526,59 @@ fn light_peer_imports_header_from_announce() { let known_stale_hash = net.peer(0).push_blocks_at(BlockId::Number(0), 1, true); import_with_announce(&mut net, &mut runtime, known_stale_hash); } + +#[test] +fn can_sync_explicit_forks() { + let _ = ::env_logger::try_init(); + let mut runtime = current_thread::Runtime::new().unwrap(); + let mut net = TestNet::new(2); + net.peer(0).push_blocks(30, false); + net.peer(1).push_blocks(30, false); + + // small fork + reorg on peer 1. + net.peer(0).push_blocks_at(BlockId::Number(30), 2, true); + let small_hash = net.peer(0).client().info().chain.best_hash; + let small_number = net.peer(0).client().info().chain.best_number; + net.peer(0).push_blocks_at(BlockId::Number(30), 10, false); + assert_eq!(net.peer(0).client().info().chain.best_number, 40); + + // peer 1 only ever had the long fork. + net.peer(1).push_blocks(10, false); + assert_eq!(net.peer(1).client().info().chain.best_number, 40); + + assert!(net.peer(0).client().header(&BlockId::Hash(small_hash)).unwrap().is_some()); + assert!(net.peer(1).client().header(&BlockId::Hash(small_hash)).unwrap().is_none()); + + // poll until the two nodes connect, otherwise announcing the block will not work + runtime.block_on(futures::future::poll_fn::<(), (), _>(|| -> Result<_, ()> { + net.poll(); + if net.peer(0).num_peers() == 0 || net.peer(1).num_peers() == 0 { + Ok(Async::NotReady) + } else { + Ok(Async::Ready(())) + } + })).unwrap(); + + // synchronization: 0 synced to longer chain and 1 didn't sync to small chain. + + assert_eq!(net.peer(0).client().info().chain.best_number, 40); + + assert!(net.peer(0).client().header(&BlockId::Hash(small_hash)).unwrap().is_some()); + assert!(!net.peer(1).client().header(&BlockId::Hash(small_hash)).unwrap().is_some()); + + // request explicit sync + let first_peer_id = net.peer(0).id(); + net.peer(1).set_sync_fork_request(vec![first_peer_id], small_hash, small_number); + + // peer 1 downloads the block. + runtime.block_on(futures::future::poll_fn::<(), (), _>(|| -> Result<_, ()> { + net.poll(); + + assert!(net.peer(0).client().header(&BlockId::Hash(small_hash)).unwrap().is_some()); + if net.peer(1).client().header(&BlockId::Hash(small_hash)).unwrap().is_none() { + return Ok(Async::NotReady) + } + Ok(Async::Ready(())) + })).unwrap(); +} +