Struct sc_network_sync::ChainSync

pub struct ChainSync<B: BlockT, Client> { /* private fields */ }

The main data structure which contains all the state for a chains active syncing strategy.

Implementations

impl<B, Client> ChainSync<B, Client>where
    Self: ChainSyncT<B>,
    B: BlockT,
    Client: HeaderBackend<B> + BlockBackend<B> + HeaderMetadata<B, Error = Error> + ProofProvider<B> + Send + Sync + 'static,

pub fn new(
    mode: SyncMode,
    client: Arc<Client>,
    protocol_id: ProtocolId,
    fork_id: &Option<String>,
    roles: Roles,
    block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
    max_parallel_downloads: u32,
    warp_sync_provider: Option<Arc<dyn WarpSyncProvider<B>>>,
    metrics_registry: Option<&Registry>,
    network_service: NetworkServiceHandle,
    import_queue: Box<dyn ImportQueueService<B>>,
    block_request_protocol_name: ProtocolName,
    state_request_protocol_name: ProtocolName,
    warp_sync_protocol_name: Option<ProtocolName>
) -> Result<(Self, ChainSyncInterfaceHandle<B>, NonDefaultSetConfig), ClientError>

Create a new instance.

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

Get config for the block announcement protocol

Examples found in repository

src/lib.rs (lines 1440-1451)

	pub fn new(
		mode: SyncMode,
		client: Arc<Client>,
		protocol_id: ProtocolId,
		fork_id: &Option<String>,
		roles: Roles,
		block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
		max_parallel_downloads: u32,
		warp_sync_provider: Option<Arc<dyn WarpSyncProvider<B>>>,
		metrics_registry: Option<&Registry>,
		network_service: service::network::NetworkServiceHandle,
		import_queue: Box<dyn ImportQueueService<B>>,
		block_request_protocol_name: ProtocolName,
		state_request_protocol_name: ProtocolName,
		warp_sync_protocol_name: Option<ProtocolName>,
	) -> Result<(Self, ChainSyncInterfaceHandle<B>, NonDefaultSetConfig), ClientError> {
		let (tx, service_rx) = tracing_unbounded("mpsc_chain_sync");
		let block_announce_config = Self::get_block_announce_proto_config(
			protocol_id,
			fork_id,
			roles,
			client.info().best_number,
			client.info().best_hash,
			client
				.block_hash(Zero::zero())
				.ok()
				.flatten()
				.expect("Genesis block exists; qed"),
		);

		let mut sync = Self {
			client,
			peers: HashMap::new(),
			blocks: BlockCollection::new(),
			best_queued_hash: Default::default(),
			best_queued_number: Zero::zero(),
			extra_justifications: ExtraRequests::new("justification"),
			mode,
			queue_blocks: Default::default(),
			fork_targets: Default::default(),
			allowed_requests: Default::default(),
			block_announce_validator,
			max_parallel_downloads,
			downloaded_blocks: 0,
			block_announce_validation: Default::default(),
			block_announce_validation_per_peer_stats: Default::default(),
			state_sync: None,
			warp_sync: None,
			warp_sync_provider,
			import_existing: false,
			gap_sync: None,
			service_rx,
			network_service,
			block_request_protocol_name,
			state_request_protocol_name,
			warp_sync_protocol_name,
			block_announce_protocol_name: block_announce_config
				.notifications_protocol
				.clone()
				.into(),
			pending_responses: Default::default(),
			import_queue,
			metrics: if let Some(r) = &metrics_registry {
				match SyncingMetrics::register(r) {
					Ok(metrics) => Some(metrics),
					Err(err) => {
						error!(target: "sync", "Failed to register metrics for ChainSync: {err:?}");
						None
					},
				}
			} else {
				None
			},
		};

		sync.reset_sync_start_point()?;
		Ok((sync, ChainSyncInterfaceHandle::new(tx), block_announce_config))
	}

pub fn on_state_response(
    &mut self,
    peer_id: PeerId,
    response: OpaqueStateResponse
) -> Option<ImportResult<B>>

Examples found in repository

src/lib.rs (line 2245)

	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
	}

pub fn on_warp_sync_response(&mut self, peer_id: PeerId, response: EncodedProof)

Examples found in repository

src/lib.rs (line 2250)

	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
	}

Trait Implementations

impl<B, Client> ChainSync<B> for ChainSync<B, Client>where
    B: BlockT,
    Client: HeaderBackend<B> + BlockBackend<B> + HeaderMetadata<B, Error = Error> + ProofProvider<B> + Send + Sync + 'static,

fn status(&self) -> SyncStatus<B>

Returns the current sync status.

fn peer_info(&self, who: &PeerId) -> Option<PeerInfo<B>>

Returns the state of the sync of the given peer.

fn num_sync_requests(&self) -> usize

Number of active forks requests. This includes requests that are pending or could be issued right away.

fn num_downloaded_blocks(&self) -> usize

Number of downloaded blocks.

fn num_peers(&self) -> usize

Returns the current number of peers stored within this state machine.

fn num_active_peers(&self) -> usize

Returns the number of peers we’re connected to and that are being queried.

fn new_peer(
    &mut self,
    who: PeerId,
    best_hash: B::Hash,
    best_number: NumberFor<B>
) -> Result<Option<BlockRequest<B>>, BadPeer>

Handle a new connected peer.

fn update_chain_info(&mut self, best_hash: &B::Hash, best_number: NumberFor<B>)

Signal that a new best block has been imported.

fn request_justification(&mut self, hash: &B::Hash, number: NumberFor<B>)

Schedule a justification request for the given block.

fn clear_justification_requests(&mut self)

Clear all pending justification requests.

fn set_sync_fork_request(
    &mut self,
    peers: Vec<PeerId>,
    hash: &B::Hash,
    number: NumberFor<B>
)

Request syncing for the given block from given set of peers.

fn on_block_data(
    &mut self,
    who: &PeerId,
    request: Option<BlockRequest<B>>,
    response: BlockResponse<B>
) -> Result<OnBlockData<B>, BadPeer>

Handle a response from the remote to a block request that we made.

fn process_block_response_data(
    &mut self,
    blocks_to_import: Result<OnBlockData<B>, BadPeer>
)

Procss received block data.

fn on_block_justification(
    &mut self,
    who: PeerId,
    response: BlockResponse<B>
) -> Result<OnBlockJustification<B>, BadPeer>

Handle a response from the remote to a justification request that we made.

fn on_justification_import(
    &mut self,
    hash: B::Hash,
    number: NumberFor<B>,
    success: bool
)

Call this when a justification has been processed by the import queue, with or without errors.

fn on_block_finalized(&mut self, hash: &B::Hash, number: NumberFor<B>)

Notify about finalization of the given block.

fn push_block_announce_validation(
    &mut self,
    who: PeerId,
    hash: B::Hash,
    announce: BlockAnnounce<B::Header>,
    is_best: bool
)

Push a block announce validation.

fn poll_block_announce_validation(
    &mut self,
    cx: &mut Context<'_>
) -> Poll<PollBlockAnnounceValidation<B::Header>>

Poll block announce validation.

fn peer_disconnected(&mut self, who: &PeerId)

Call when a peer has disconnected. Canceled obsolete block request may result in some blocks being ready for import, so this functions checks for such blocks and returns them.

fn metrics(&self) -> Metrics

Return some key metrics.

fn block_response_into_blocks(
    &self,
    request: &BlockRequest<B>,
    response: OpaqueBlockResponse
) -> Result<Vec<BlockData<B>>, String>

Access blocks from implementation-specific block response.

fn poll(
    &mut self,
    cx: &mut Context<'_>
) -> Poll<PollBlockAnnounceValidation<B::Header>>

Advance the state of ChainSync

fn send_block_request(&mut self, who: PeerId, request: BlockRequest<B>)

Send block request to peer