Enum sc_network_sync::PeerSyncState

pub enum PeerSyncState<B: BlockT> {
    Available,
    AncestorSearch {
        start: NumberFor<B>,
        current: NumberFor<B>,
        state: AncestorSearchState<B>,
    },
    DownloadingNew(NumberFor<B>),
    DownloadingStale(B::Hash),
    DownloadingJustification(B::Hash),
    DownloadingState,
    DownloadingWarpProof,
    DownloadingWarpTargetBlock,
    DownloadingGap(NumberFor<B>),
}

The state of syncing between a Peer and ourselves.

Generally two categories, “busy” or Available. If busy, the enum defines what we are busy with.

Variants

Available

Available for sync requests.

AncestorSearch

Fields

start: NumberFor<B>

current: NumberFor<B>

state: AncestorSearchState<B>

Searching for ancestors the Peer has in common with us.

DownloadingNew(NumberFor<B>)

Actively downloading new blocks, starting from the given Number.

DownloadingStale(B::Hash)

Downloading a stale block with given Hash. Stale means that it is a block with a number that is lower than our best number. It might be from a fork and not necessarily already imported.

DownloadingJustification(B::Hash)

Downloading justification for given block hash.

DownloadingState

Downloading state.

DownloadingWarpProof

Downloading warp proof.

DownloadingWarpTargetBlock

Downloading warp sync target block.

DownloadingGap(NumberFor<B>)

Actively downloading block history after warp sync.

Implementations

impl<B: BlockT> PeerSyncState<B>

pub fn is_available(&self) -> bool

Examples found in repository

src/lib.rs (line 1959)

	fn warp_target_block_request(&mut self) -> Option<(PeerId, BlockRequest<B>)> {
		let sync = &self.warp_sync.as_ref()?;

		if self.allowed_requests.is_empty() ||
			sync.is_complete() ||
			self.peers
				.iter()
				.any(|(_, peer)| peer.state == PeerSyncState::DownloadingWarpTargetBlock)
		{
			// Only one pending warp target block request is allowed.
			return None
		}

		if let Some((target_number, request)) = sync.next_target_block_request() {
			// Find a random peer that has a block with the target number.
			for (id, peer) in self.peers.iter_mut() {
				if peer.state.is_available() && peer.best_number >= target_number {
					trace!(target: "sync", "New warp target block request for {}", id);
					peer.state = PeerSyncState::DownloadingWarpTargetBlock;
					self.allowed_requests.clear();
					return Some((*id, request))
				}
			}
		}

		None
	}

	/// Get config for the block announcement protocol
	pub fn get_block_announce_proto_config(
		protocol_id: ProtocolId,
		fork_id: &Option<String>,
		roles: Roles,
		best_number: NumberFor<B>,
		best_hash: B::Hash,
		genesis_hash: B::Hash,
	) -> NonDefaultSetConfig {
		let block_announces_protocol = {
			let genesis_hash = genesis_hash.as_ref();
			if let Some(ref fork_id) = fork_id {
				format!(
					"/{}/{}/block-announces/1",
					array_bytes::bytes2hex("", genesis_hash),
					fork_id
				)
			} else {
				format!("/{}/block-announces/1", array_bytes::bytes2hex("", genesis_hash))
			}
		};

		NonDefaultSetConfig {
			notifications_protocol: block_announces_protocol.into(),
			fallback_names: iter::once(
				format!("/{}/block-announces/1", protocol_id.as_ref()).into(),
			)
			.collect(),
			max_notification_size: MAX_BLOCK_ANNOUNCE_SIZE,
			handshake: Some(NotificationHandshake::new(BlockAnnouncesHandshake::<B>::build(
				roles,
				best_number,
				best_hash,
				genesis_hash,
			))),
			// NOTE: `set_config` will be ignored by `protocol.rs` as the block announcement
			// protocol is still hardcoded into the peerset.
			set_config: SetConfig {
				in_peers: 0,
				out_peers: 0,
				reserved_nodes: Vec::new(),
				non_reserved_mode: NonReservedPeerMode::Deny,
			},
		}
	}

	fn decode_block_response(response: &[u8]) -> Result<OpaqueBlockResponse, String> {
		let response = schema::v1::BlockResponse::decode(response)
			.map_err(|error| format!("Failed to decode block response: {error}"))?;

		Ok(OpaqueBlockResponse(Box::new(response)))
	}

	fn decode_state_response(response: &[u8]) -> Result<OpaqueStateResponse, String> {
		let response = StateResponse::decode(response)
			.map_err(|error| format!("Failed to decode state response: {error}"))?;

		Ok(OpaqueStateResponse(Box::new(response)))
	}

	fn send_state_request(&mut self, who: PeerId, request: OpaqueStateRequest) {
		let (tx, rx) = oneshot::channel();

		if self.peers.contains_key(&who) {
			self.pending_responses
				.push(Box::pin(async move { (who, PeerRequest::State, rx.await) }));
		}

		match self.encode_state_request(&request) {
			Ok(data) => {
				self.network_service.start_request(
					who,
					self.state_request_protocol_name.clone(),
					data,
					tx,
					IfDisconnected::ImmediateError,
				);
			},
			Err(err) => {
				log::warn!(
					target: "sync",
					"Failed to encode state request {:?}: {:?}",
					request, err
				);
			},
		}
	}

	fn send_warp_sync_request(&mut self, who: PeerId, request: WarpProofRequest<B>) {
		let (tx, rx) = oneshot::channel();

		if self.peers.contains_key(&who) {
			self.pending_responses
				.push(Box::pin(async move { (who, PeerRequest::WarpProof, rx.await) }));
		}

		match &self.warp_sync_protocol_name {
			Some(name) => self.network_service.start_request(
				who,
				name.clone(),
				request.encode(),
				tx,
				IfDisconnected::ImmediateError,
			),
			None => {
				log::warn!(
					target: "sync",
					"Trying to send warp sync request when no protocol is configured {:?}",
					request,
				);
			},
		}
	}

	fn on_block_response(
		&mut self,
		peer_id: PeerId,
		request: BlockRequest<B>,
		response: OpaqueBlockResponse,
	) -> Option<ImportResult<B>> {
		let blocks = match self.block_response_into_blocks(&request, response) {
			Ok(blocks) => blocks,
			Err(err) => {
				debug!(target: "sync", "Failed to decode block response from {}: {}", peer_id, err);
				self.network_service.report_peer(peer_id, rep::BAD_MESSAGE);
				return None
			},
		};

		let block_response = BlockResponse::<B> { id: request.id, blocks };

		let blocks_range = || match (
			block_response
				.blocks
				.first()
				.and_then(|b| b.header.as_ref().map(|h| h.number())),
			block_response.blocks.last().and_then(|b| b.header.as_ref().map(|h| h.number())),
		) {
			(Some(first), Some(last)) if first != last => format!(" ({}..{})", first, last),
			(Some(first), Some(_)) => format!(" ({})", first),
			_ => Default::default(),
		};
		trace!(
			target: "sync",
			"BlockResponse {} from {} with {} blocks {}",
			block_response.id,
			peer_id,
			block_response.blocks.len(),
			blocks_range(),
		);

		if request.fields == BlockAttributes::JUSTIFICATION {
			match self.on_block_justification(peer_id, block_response) {
				Ok(OnBlockJustification::Nothing) => None,
				Ok(OnBlockJustification::Import { peer, hash, number, justifications }) => {
					self.import_justifications(peer, hash, number, justifications);
					None
				},
				Err(BadPeer(id, repu)) => {
					self.network_service
						.disconnect_peer(id, self.block_announce_protocol_name.clone());
					self.network_service.report_peer(id, repu);
					None
				},
			}
		} else {
			match self.on_block_data(&peer_id, Some(request), block_response) {
				Ok(OnBlockData::Import(origin, blocks)) => {
					self.import_blocks(origin, blocks);
					None
				},
				Ok(OnBlockData::Request(peer, req)) => {
					self.send_block_request(peer, req);
					None
				},
				Ok(OnBlockData::Continue) => None,
				Err(BadPeer(id, repu)) => {
					self.network_service
						.disconnect_peer(id, self.block_announce_protocol_name.clone());
					self.network_service.report_peer(id, repu);
					None
				},
			}
		}
	}

	pub fn on_state_response(
		&mut self,
		peer_id: PeerId,
		response: OpaqueStateResponse,
	) -> Option<ImportResult<B>> {
		match self.on_state_data(&peer_id, response) {
			Ok(OnStateData::Import(origin, block)) =>
				Some(ImportResult::BlockImport(origin, vec![block])),
			Ok(OnStateData::Continue) => None,
			Err(BadPeer(id, repu)) => {
				self.network_service
					.disconnect_peer(id, self.block_announce_protocol_name.clone());
				self.network_service.report_peer(id, repu);
				None
			},
		}
	}

	pub fn on_warp_sync_response(&mut self, peer_id: PeerId, response: EncodedProof) {
		if let Err(BadPeer(id, repu)) = self.on_warp_sync_data(&peer_id, response) {
			self.network_service
				.disconnect_peer(id, self.block_announce_protocol_name.clone());
			self.network_service.report_peer(id, repu);
		}
	}

	fn process_outbound_requests(&mut self) {
		for (id, request) in self.block_requests() {
			self.send_block_request(id, request);
		}

		if let Some((id, request)) = self.state_request() {
			self.send_state_request(id, request);
		}

		for (id, request) in self.justification_requests().collect::<Vec<_>>() {
			self.send_block_request(id, request);
		}

		if let Some((id, request)) = self.warp_sync_request() {
			self.send_warp_sync_request(id, request);
		}
	}

	fn poll_pending_responses(&mut self, cx: &mut std::task::Context) -> Poll<ImportResult<B>> {
		while let Poll::Ready(Some((id, request, response))) =
			self.pending_responses.poll_next_unpin(cx)
		{
			match response {
				Ok(Ok(resp)) => match request {
					PeerRequest::Block(req) => {
						let response = match Self::decode_block_response(&resp[..]) {
							Ok(proto) => proto,
							Err(e) => {
								debug!(
									target: "sync",
									"Failed to decode block response from peer {:?}: {:?}.",
									id,
									e
								);
								self.network_service.report_peer(id, rep::BAD_MESSAGE);
								self.network_service
									.disconnect_peer(id, self.block_announce_protocol_name.clone());
								continue
							},
						};

						if let Some(import) = self.on_block_response(id, req, response) {
							return Poll::Ready(import)
						}
					},
					PeerRequest::State => {
						let response = match Self::decode_state_response(&resp[..]) {
							Ok(proto) => proto,
							Err(e) => {
								debug!(
									target: "sync",
									"Failed to decode state response from peer {:?}: {:?}.",
									id,
									e
								);
								self.network_service.report_peer(id, rep::BAD_MESSAGE);
								self.network_service
									.disconnect_peer(id, self.block_announce_protocol_name.clone());
								continue
							},
						};

						if let Some(import) = self.on_state_response(id, response) {
							return Poll::Ready(import)
						}
					},
					PeerRequest::WarpProof => {
						self.on_warp_sync_response(id, EncodedProof(resp));
					},
				},
				Ok(Err(e)) => {
					debug!(target: "sync", "Request to peer {:?} failed: {:?}.", id, e);

					match e {
						RequestFailure::Network(OutboundFailure::Timeout) => {
							self.network_service.report_peer(id, rep::TIMEOUT);
							self.network_service
								.disconnect_peer(id, self.block_announce_protocol_name.clone());
						},
						RequestFailure::Network(OutboundFailure::UnsupportedProtocols) => {
							self.network_service.report_peer(id, rep::BAD_PROTOCOL);
							self.network_service
								.disconnect_peer(id, self.block_announce_protocol_name.clone());
						},
						RequestFailure::Network(OutboundFailure::DialFailure) => {
							self.network_service
								.disconnect_peer(id, self.block_announce_protocol_name.clone());
						},
						RequestFailure::Refused => {
							self.network_service.report_peer(id, rep::REFUSED);
							self.network_service
								.disconnect_peer(id, self.block_announce_protocol_name.clone());
						},
						RequestFailure::Network(OutboundFailure::ConnectionClosed) |
						RequestFailure::NotConnected => {
							self.network_service
								.disconnect_peer(id, self.block_announce_protocol_name.clone());
						},
						RequestFailure::UnknownProtocol => {
							debug_assert!(false, "Block request protocol should always be known.");
						},
						RequestFailure::Obsolete => {
							debug_assert!(
								false,
								"Can not receive `RequestFailure::Obsolete` after dropping the \
								 response receiver.",
							);
						},
					}
				},
				Err(oneshot::Canceled) => {
					trace!(
						target: "sync",
						"Request to peer {:?} failed due to oneshot being canceled.",
						id,
					);
					self.network_service
						.disconnect_peer(id, self.block_announce_protocol_name.clone());
				},
			}
		}

		Poll::Pending
	}

	/// Create implementation-specific block request.
	fn create_opaque_block_request(&self, request: &BlockRequest<B>) -> OpaqueBlockRequest {
		OpaqueBlockRequest(Box::new(schema::v1::BlockRequest {
			fields: request.fields.to_be_u32(),
			from_block: match request.from {
				FromBlock::Hash(h) => Some(schema::v1::block_request::FromBlock::Hash(h.encode())),
				FromBlock::Number(n) =>
					Some(schema::v1::block_request::FromBlock::Number(n.encode())),
			},
			direction: request.direction as i32,
			max_blocks: request.max.unwrap_or(0),
			support_multiple_justifications: true,
		}))
	}

	fn encode_block_request(&self, request: &OpaqueBlockRequest) -> Result<Vec<u8>, String> {
		let request: &schema::v1::BlockRequest = request.0.downcast_ref().ok_or_else(|| {
			"Failed to downcast opaque block response during encoding, this is an \
				implementation bug."
				.to_string()
		})?;

		Ok(request.encode_to_vec())
	}

	fn encode_state_request(&self, request: &OpaqueStateRequest) -> Result<Vec<u8>, String> {
		let request: &StateRequest = request.0.downcast_ref().ok_or_else(|| {
			"Failed to downcast opaque state response during encoding, this is an \
				implementation bug."
				.to_string()
		})?;

		Ok(request.encode_to_vec())
	}

	fn justification_requests<'a>(
		&'a mut self,
	) -> Box<dyn Iterator<Item = (PeerId, BlockRequest<B>)> + 'a> {
		let peers = &mut self.peers;
		let mut matcher = self.extra_justifications.matcher();
		Box::new(std::iter::from_fn(move || {
			if let Some((peer, request)) = matcher.next(peers) {
				peers
					.get_mut(&peer)
					.expect(
						"`Matcher::next` guarantees the `PeerId` comes from the given peers; qed",
					)
					.state = PeerSyncState::DownloadingJustification(request.0);
				let req = BlockRequest::<B> {
					id: 0,
					fields: BlockAttributes::JUSTIFICATION,
					from: FromBlock::Hash(request.0),
					direction: Direction::Ascending,
					max: Some(1),
				};
				Some((peer, req))
			} else {
				None
			}
		}))
	}

	fn block_requests(&mut self) -> Vec<(PeerId, BlockRequest<B>)> {
		if self.mode == SyncMode::Warp {
			return self
				.warp_target_block_request()
				.map_or_else(|| Vec::new(), |req| Vec::from([req]))
		}

		if self.allowed_requests.is_empty() || self.state_sync.is_some() {
			return Vec::new()
		}

		if self.queue_blocks.len() > MAX_IMPORTING_BLOCKS {
			trace!(target: "sync", "Too many blocks in the queue.");
			return Vec::new()
		}
		let is_major_syncing = self.status().state.is_major_syncing();
		let attrs = self.required_block_attributes();
		let blocks = &mut self.blocks;
		let fork_targets = &mut self.fork_targets;
		let last_finalized =
			std::cmp::min(self.best_queued_number, self.client.info().finalized_number);
		let best_queued = self.best_queued_number;
		let client = &self.client;
		let queue = &self.queue_blocks;
		let allowed_requests = self.allowed_requests.take();
		let max_parallel = if is_major_syncing { 1 } else { self.max_parallel_downloads };
		let gap_sync = &mut self.gap_sync;
		self.peers
			.iter_mut()
			.filter_map(move |(&id, peer)| {
				if !peer.state.is_available() || !allowed_requests.contains(&id) {
					return None
				}

				// If our best queued is more than `MAX_BLOCKS_TO_LOOK_BACKWARDS` blocks away from
				// the common number, the peer best number is higher than our best queued and the
				// common number is smaller than the last finalized block number, we should do an
				// ancestor search to find a better common block. If the queue is full we wait till
				// all blocks are imported though.
				if best_queued.saturating_sub(peer.common_number) >
					MAX_BLOCKS_TO_LOOK_BACKWARDS.into() &&
					best_queued < peer.best_number &&
					peer.common_number < last_finalized &&
					queue.len() <= MAJOR_SYNC_BLOCKS.into()
				{
					trace!(
						target: "sync",
						"Peer {:?} common block {} too far behind of our best {}. Starting ancestry search.",
						id,
						peer.common_number,
						best_queued,
					);
					let current = std::cmp::min(peer.best_number, best_queued);
					peer.state = PeerSyncState::AncestorSearch {
						current,
						start: best_queued,
						state: AncestorSearchState::ExponentialBackoff(One::one()),
					};
					Some((id, ancestry_request::<B>(current)))
				} else if let Some((range, req)) = peer_block_request(
					&id,
					peer,
					blocks,
					attrs,
					max_parallel,
					last_finalized,
					best_queued,
				) {
					peer.state = PeerSyncState::DownloadingNew(range.start);
					trace!(
						target: "sync",
						"New block request for {}, (best:{}, common:{}) {:?}",
						id,
						peer.best_number,
						peer.common_number,
						req,
					);
					Some((id, req))
				} else if let Some((hash, req)) = fork_sync_request(
					&id,
					fork_targets,
					best_queued,
					last_finalized,
					attrs,
					|hash| {
						if queue.contains(hash) {
							BlockStatus::Queued
						} else {
							client
								.block_status(&BlockId::Hash(*hash))
								.unwrap_or(BlockStatus::Unknown)
						}
					},
				) {
					trace!(target: "sync", "Downloading fork {:?} from {}", hash, id);
					peer.state = PeerSyncState::DownloadingStale(hash);
					Some((id, req))
				} else if let Some((range, req)) = gap_sync.as_mut().and_then(|sync| {
					peer_gap_block_request(
						&id,
						peer,
						&mut sync.blocks,
						attrs,
						sync.target,
						sync.best_queued_number,
					)
				}) {
					peer.state = PeerSyncState::DownloadingGap(range.start);
					trace!(
						target: "sync",
						"New gap block request for {}, (best:{}, common:{}) {:?}",
						id,
						peer.best_number,
						peer.common_number,
						req,
					);
					Some((id, req))
				} else {
					None
				}
			})
			.collect()
		// Box::new(iter)
	}

	fn state_request(&mut self) -> Option<(PeerId, OpaqueStateRequest)> {
		if self.allowed_requests.is_empty() {
			return None
		}
		if (self.state_sync.is_some() || self.warp_sync.is_some()) &&
			self.peers.iter().any(|(_, peer)| peer.state == PeerSyncState::DownloadingState)
		{
			// Only one pending state request is allowed.
			return None
		}
		if let Some(sync) = &self.state_sync {
			if sync.is_complete() {
				return None
			}

			for (id, peer) in self.peers.iter_mut() {
				if peer.state.is_available() && peer.common_number >= sync.target_block_num() {
					peer.state = PeerSyncState::DownloadingState;
					let request = sync.next_request();
					trace!(target: "sync", "New StateRequest for {}: {:?}", id, request);
					self.allowed_requests.clear();
					return Some((*id, OpaqueStateRequest(Box::new(request))))
				}
			}
		}
		if let Some(sync) = &self.warp_sync {
			if sync.is_complete() {
				return None
			}
			if let (Some(request), Some(target)) =
				(sync.next_state_request(), sync.target_block_number())
			{
				for (id, peer) in self.peers.iter_mut() {
					if peer.state.is_available() && peer.best_number >= target {
						trace!(target: "sync", "New StateRequest for {}: {:?}", id, request);
						peer.state = PeerSyncState::DownloadingState;
						self.allowed_requests.clear();
						return Some((*id, OpaqueStateRequest(Box::new(request))))
					}
				}
			}
		}
		None
	}

	fn warp_sync_request(&mut self) -> Option<(PeerId, WarpProofRequest<B>)> {
		if let Some(sync) = &self.warp_sync {
			if self.allowed_requests.is_empty() ||
				sync.is_complete() ||
				self.peers
					.iter()
					.any(|(_, peer)| peer.state == PeerSyncState::DownloadingWarpProof)
			{
				// Only one pending state request is allowed.
				return None
			}
			if let Some(request) = sync.next_warp_proof_request() {
				let mut targets: Vec<_> = self.peers.values().map(|p| p.best_number).collect();
				if !targets.is_empty() {
					targets.sort();
					let median = targets[targets.len() / 2];
					// Find a random peer that is synced as much as peer majority.
					for (id, peer) in self.peers.iter_mut() {
						if peer.state.is_available() && peer.best_number >= median {
							trace!(target: "sync", "New WarpProofRequest for {}", id);
							peer.state = PeerSyncState::DownloadingWarpProof;
							self.allowed_requests.clear();
							return Some((*id, request))
						}
					}
				}
			}
		}
		None
	}

Trait Implementations

impl<B: Clone + BlockT> Clone for PeerSyncState<B>where
    B::Hash: Clone,

fn clone(&self) -> PeerSyncState<B>

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<B: Debug + BlockT> Debug for PeerSyncState<B>where
    B::Hash: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<B: PartialEq + BlockT> PartialEq<PeerSyncState<B>> for PeerSyncState<B>where
    B::Hash: PartialEq,

fn eq(&self, other: &PeerSyncState<B>) -> bool

This method tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

This method tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl<B: Copy + BlockT> Copy for PeerSyncState<B>where
    B::Hash: Copy,

impl<B: Eq + BlockT> Eq for PeerSyncState<B>where
    B::Hash: Eq,

impl<B: BlockT> StructuralEq for PeerSyncState<B>

impl<B: BlockT> StructuralPartialEq for PeerSyncState<B>