1use crate::{
23 block_announce_validator::{
24 BlockAnnounceValidationResult, BlockAnnounceValidator as BlockAnnounceValidatorStream,
25 },
26 pending_responses::{PendingResponses, ResponseEvent},
27 service::{
28 self,
29 syncing_service::{SyncingService, ToServiceCommand},
30 },
31 strategy::{SyncingAction, SyncingStrategy},
32 types::{BadPeer, ExtendedPeerInfo, SyncEvent},
33 LOG_TARGET,
34};
35
36use codec::{Decode, DecodeAll, Encode};
37use futures::{channel::oneshot, StreamExt};
38use log::{debug, error, trace, warn};
39use prometheus_endpoint::{
40 register, Counter, Gauge, MetricSource, Opts, PrometheusError, Registry, SourcedGauge, U64,
41};
42use schnellru::{ByLength, LruMap};
43use tokio::time::{Interval, MissedTickBehavior};
44
45use sc_client_api::{BlockBackend, HeaderBackend, ProofProvider};
46use sc_consensus::{import_queue::ImportQueueService, IncomingBlock};
47use sc_network::{
48 config::{FullNetworkConfiguration, NotificationHandshake, ProtocolId, SetConfig},
49 peer_store::PeerStoreProvider,
50 request_responses::{OutboundFailure, RequestFailure},
51 service::{
52 traits::{Direction, NotificationConfig, NotificationEvent, ValidationResult},
53 NotificationMetrics,
54 },
55 types::ProtocolName,
56 utils::LruHashSet,
57 NetworkBackend, NotificationService, ReputationChange,
58};
59use sc_network_common::{
60 role::Roles,
61 sync::message::{BlockAnnounce, BlockAnnouncesHandshake, BlockState},
62};
63use sc_network_types::PeerId;
64use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
65use sp_blockchain::{Error as ClientError, HeaderMetadata};
66use sp_consensus::{block_validation::BlockAnnounceValidator, BlockOrigin};
67use sp_runtime::{
68 traits::{Block as BlockT, Header, NumberFor, Zero},
69 Justifications,
70};
71
72use std::{
73 collections::{HashMap, HashSet},
74 iter,
75 num::NonZeroUsize,
76 sync::{
77 atomic::{AtomicBool, AtomicUsize, Ordering},
78 Arc,
79 },
80};
81
82const TICK_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(1100);
84
85const MAX_KNOWN_BLOCKS: usize = 1024; const MAX_BLOCK_ANNOUNCE_SIZE: u64 = 1024 * 1024;
90
91mod rep {
92 use sc_network::ReputationChange as Rep;
93 pub const GENESIS_MISMATCH: Rep = Rep::new_fatal("Genesis mismatch");
95 pub const BAD_BLOCK_ANNOUNCEMENT: Rep = Rep::new(-(1 << 12), "Bad block announcement");
97 pub const BAD_PROTOCOL: Rep = Rep::new_fatal("Unsupported protocol");
99 pub const REFUSED: Rep = Rep::new(-(1 << 10), "Request refused");
101 pub const TIMEOUT: Rep = Rep::new(-(1 << 10), "Request timeout");
103 pub const IO: Rep = Rep::new(-(1 << 10), "IO error during request");
105}
106
107struct Metrics {
108 peers: Gauge<U64>,
109 import_queue_blocks_submitted: Counter<U64>,
110 import_queue_justifications_submitted: Counter<U64>,
111}
112
113impl Metrics {
114 fn register(r: &Registry, major_syncing: Arc<AtomicBool>) -> Result<Self, PrometheusError> {
115 MajorSyncingGauge::register(r, major_syncing)?;
116 Ok(Self {
117 peers: {
118 let g = Gauge::new("substrate_sync_peers", "Number of peers we sync with")?;
119 register(g, r)?
120 },
121 import_queue_blocks_submitted: {
122 let c = Counter::new(
123 "substrate_sync_import_queue_blocks_submitted",
124 "Number of blocks submitted to the import queue.",
125 )?;
126 register(c, r)?
127 },
128 import_queue_justifications_submitted: {
129 let c = Counter::new(
130 "substrate_sync_import_queue_justifications_submitted",
131 "Number of justifications submitted to the import queue.",
132 )?;
133 register(c, r)?
134 },
135 })
136 }
137}
138
139#[derive(Clone)]
141pub struct MajorSyncingGauge(Arc<AtomicBool>);
142
143impl MajorSyncingGauge {
144 fn register(registry: &Registry, value: Arc<AtomicBool>) -> Result<(), PrometheusError> {
147 prometheus_endpoint::register(
148 SourcedGauge::new(
149 &Opts::new(
150 "substrate_sub_libp2p_is_major_syncing",
151 "Whether the node is performing a major sync or not.",
152 ),
153 MajorSyncingGauge(value),
154 )?,
155 registry,
156 )?;
157
158 Ok(())
159 }
160}
161
162impl MetricSource for MajorSyncingGauge {
163 type N = u64;
164
165 fn collect(&self, mut set: impl FnMut(&[&str], Self::N)) {
166 set(&[], self.0.load(Ordering::Relaxed) as u64);
167 }
168}
169
170#[derive(Debug)]
172pub struct Peer<B: BlockT> {
173 pub info: ExtendedPeerInfo<B>,
174 pub known_blocks: LruHashSet<B::Hash>,
176 inbound: bool,
178}
179
180pub struct SyncingEngine<B: BlockT, Client> {
181 strategy: Box<dyn SyncingStrategy<B>>,
183
184 client: Arc<Client>,
186
187 num_connected: Arc<AtomicUsize>,
189
190 is_major_syncing: Arc<AtomicBool>,
192
193 network_service: service::network::NetworkServiceHandle,
195
196 service_rx: TracingUnboundedReceiver<ToServiceCommand<B>>,
198
199 roles: Roles,
201
202 genesis_hash: B::Hash,
204
205 event_streams: Vec<TracingUnboundedSender<SyncEvent>>,
207
208 tick_timeout: Interval,
210
211 peers: HashMap<PeerId, Peer<B>>,
213
214 important_peers: HashSet<PeerId>,
217
218 default_peers_set_no_slot_connected_peers: HashSet<PeerId>,
220
221 default_peers_set_no_slot_peers: HashSet<PeerId>,
223
224 default_peers_set_num_full: usize,
227
228 default_peers_set_num_light: usize,
230
231 max_in_peers: usize,
233
234 num_in_peers: usize,
236
237 block_announce_validator: BlockAnnounceValidatorStream<B>,
239
240 block_announce_data_cache: LruMap<B::Hash, Vec<u8>>,
242
243 boot_node_ids: HashSet<PeerId>,
245
246 block_announce_protocol_name: ProtocolName,
248
249 metrics: Option<Metrics>,
251
252 notification_service: Box<dyn NotificationService>,
254
255 peer_store_handle: Arc<dyn PeerStoreProvider>,
257
258 pending_responses: PendingResponses,
260
261 import_queue: Box<dyn ImportQueueService<B>>,
263}
264
265impl<B: BlockT, Client> SyncingEngine<B, Client>
266where
267 B: BlockT,
268 Client: HeaderBackend<B>
269 + BlockBackend<B>
270 + HeaderMetadata<B, Error = sp_blockchain::Error>
271 + ProofProvider<B>
272 + Send
273 + Sync
274 + 'static,
275{
276 pub fn new<N>(
277 roles: Roles,
278 client: Arc<Client>,
279 metrics_registry: Option<&Registry>,
280 network_metrics: NotificationMetrics,
281 net_config: &FullNetworkConfiguration<B, <B as BlockT>::Hash, N>,
282 protocol_id: ProtocolId,
283 fork_id: Option<&str>,
284 block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
285 syncing_strategy: Box<dyn SyncingStrategy<B>>,
286 network_service: service::network::NetworkServiceHandle,
287 import_queue: Box<dyn ImportQueueService<B>>,
288 peer_store_handle: Arc<dyn PeerStoreProvider>,
289 ) -> Result<(Self, SyncingService<B>, N::NotificationProtocolConfig), ClientError>
290 where
291 N: NetworkBackend<B, <B as BlockT>::Hash>,
292 {
293 let cache_capacity = (net_config.network_config.default_peers_set.in_peers +
294 net_config.network_config.default_peers_set.out_peers)
295 .max(1);
296 let important_peers = {
297 let mut imp_p = HashSet::new();
298 for reserved in &net_config.network_config.default_peers_set.reserved_nodes {
299 imp_p.insert(reserved.peer_id);
300 }
301 for config in net_config.notification_protocols() {
302 let peer_ids = config.set_config().reserved_nodes.iter().map(|info| info.peer_id);
303 imp_p.extend(peer_ids);
304 }
305
306 imp_p.shrink_to_fit();
307 imp_p
308 };
309 let boot_node_ids = {
310 let mut list = HashSet::new();
311 for node in &net_config.network_config.boot_nodes {
312 list.insert(node.peer_id);
313 }
314 list.shrink_to_fit();
315 list
316 };
317 let default_peers_set_no_slot_peers = {
318 let mut no_slot_p: HashSet<PeerId> = net_config
319 .network_config
320 .default_peers_set
321 .reserved_nodes
322 .iter()
323 .map(|reserved| reserved.peer_id)
324 .collect();
325 no_slot_p.shrink_to_fit();
326 no_slot_p
327 };
328 let default_peers_set_num_full =
329 net_config.network_config.default_peers_set_num_full as usize;
330 let default_peers_set_num_light = {
331 let total = net_config.network_config.default_peers_set.out_peers +
332 net_config.network_config.default_peers_set.in_peers;
333 total.saturating_sub(net_config.network_config.default_peers_set_num_full) as usize
334 };
335
336 let info = client.info();
337
338 let (block_announce_config, notification_service) =
339 Self::get_block_announce_proto_config::<N>(
340 protocol_id,
341 fork_id,
342 roles,
343 info.best_number,
344 info.best_hash,
345 info.genesis_hash,
346 &net_config.network_config.default_peers_set,
347 network_metrics,
348 Arc::clone(&peer_store_handle),
349 );
350
351 let block_announce_protocol_name = block_announce_config.protocol_name().clone();
352 let (tx, service_rx) = tracing_unbounded("mpsc_chain_sync", 100_000);
353 let num_connected = Arc::new(AtomicUsize::new(0));
354 let is_major_syncing = Arc::new(AtomicBool::new(false));
355
356 let max_full_peers = net_config.network_config.default_peers_set_num_full;
359 let max_out_peers = net_config.network_config.default_peers_set.out_peers;
360 let max_in_peers = (max_full_peers - max_out_peers) as usize;
361
362 let tick_timeout = {
363 let mut interval = tokio::time::interval(TICK_TIMEOUT);
364 interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
365 interval
366 };
367
368 Ok((
369 Self {
370 roles,
371 client,
372 strategy: syncing_strategy,
373 network_service,
374 peers: HashMap::new(),
375 block_announce_data_cache: LruMap::new(ByLength::new(cache_capacity)),
376 block_announce_protocol_name,
377 block_announce_validator: BlockAnnounceValidatorStream::new(
378 block_announce_validator,
379 ),
380 num_connected: num_connected.clone(),
381 is_major_syncing: is_major_syncing.clone(),
382 service_rx,
383 genesis_hash: info.genesis_hash,
384 important_peers,
385 default_peers_set_no_slot_connected_peers: HashSet::new(),
386 boot_node_ids,
387 default_peers_set_no_slot_peers,
388 default_peers_set_num_full,
389 default_peers_set_num_light,
390 num_in_peers: 0usize,
391 max_in_peers,
392 event_streams: Vec::new(),
393 notification_service,
394 tick_timeout,
395 peer_store_handle,
396 metrics: if let Some(r) = metrics_registry {
397 match Metrics::register(r, is_major_syncing.clone()) {
398 Ok(metrics) => Some(metrics),
399 Err(err) => {
400 log::error!(target: LOG_TARGET, "Failed to register metrics {err:?}");
401 None
402 },
403 }
404 } else {
405 None
406 },
407 pending_responses: PendingResponses::new(),
408 import_queue,
409 },
410 SyncingService::new(tx, num_connected, is_major_syncing),
411 block_announce_config,
412 ))
413 }
414
415 fn update_peer_info(
416 &mut self,
417 peer_id: &PeerId,
418 best_hash: B::Hash,
419 best_number: NumberFor<B>,
420 ) {
421 if let Some(ref mut peer) = self.peers.get_mut(peer_id) {
422 peer.info.best_hash = best_hash;
423 peer.info.best_number = best_number;
424 }
425 }
426
427 fn process_block_announce_validation_result(
429 &mut self,
430 validation_result: BlockAnnounceValidationResult<B::Header>,
431 ) {
432 match validation_result {
433 BlockAnnounceValidationResult::Skip { peer_id: _ } => {},
434 BlockAnnounceValidationResult::Process { is_new_best, peer_id, announce } => {
435 if let Some((best_hash, best_number)) =
436 self.strategy.on_validated_block_announce(is_new_best, peer_id, &announce)
437 {
438 self.update_peer_info(&peer_id, best_hash, best_number);
439 }
440
441 if let Some(data) = announce.data {
442 if !data.is_empty() {
443 self.block_announce_data_cache.insert(announce.header.hash(), data);
444 }
445 }
446 },
447 BlockAnnounceValidationResult::Failure { peer_id, disconnect } => {
448 if disconnect {
449 log::debug!(
450 target: LOG_TARGET,
451 "Disconnecting peer {peer_id} due to block announce validation failure",
452 );
453 self.network_service
454 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
455 }
456
457 self.network_service.report_peer(peer_id, rep::BAD_BLOCK_ANNOUNCEMENT);
458 },
459 }
460 }
461
462 pub fn push_block_announce_validation(
464 &mut self,
465 peer_id: PeerId,
466 announce: BlockAnnounce<B::Header>,
467 ) {
468 let hash = announce.header.hash();
469
470 let peer = match self.peers.get_mut(&peer_id) {
471 Some(p) => p,
472 None => {
473 log::error!(
474 target: LOG_TARGET,
475 "Received block announce from disconnected peer {peer_id}",
476 );
477 debug_assert!(false);
478 return;
479 },
480 };
481 peer.known_blocks.insert(hash);
482
483 if peer.info.roles.is_full() {
484 let is_best = match announce.state.unwrap_or(BlockState::Best) {
485 BlockState::Best => true,
486 BlockState::Normal => false,
487 };
488
489 self.block_announce_validator
490 .push_block_announce_validation(peer_id, hash, announce, is_best);
491 }
492 }
493
494 pub fn announce_block(&mut self, hash: B::Hash, data: Option<Vec<u8>>) {
499 let header = match self.client.header(hash) {
500 Ok(Some(header)) => header,
501 Ok(None) => {
502 log::warn!(target: LOG_TARGET, "Trying to announce unknown block: {hash}");
503 return;
504 },
505 Err(e) => {
506 log::warn!(target: LOG_TARGET, "Error reading block header {hash}: {e}");
507 return;
508 },
509 };
510
511 if header.number().is_zero() {
513 return;
514 }
515
516 let is_best = self.client.info().best_hash == hash;
517 log::debug!(target: LOG_TARGET, "Reannouncing block {hash:?} is_best: {is_best}");
518
519 let data = data
520 .or_else(|| self.block_announce_data_cache.get(&hash).cloned())
521 .unwrap_or_default();
522
523 for (peer_id, ref mut peer) in self.peers.iter_mut() {
524 let inserted = peer.known_blocks.insert(hash);
525 if inserted {
526 log::trace!(target: LOG_TARGET, "Announcing block {hash:?} to {peer_id}");
527 let message = BlockAnnounce {
528 header: header.clone(),
529 state: if is_best { Some(BlockState::Best) } else { Some(BlockState::Normal) },
530 data: Some(data.clone()),
531 };
532
533 let _ = self.notification_service.send_sync_notification(peer_id, message.encode());
534 }
535 }
536 }
537
538 pub async fn run(mut self) {
539 loop {
540 tokio::select! {
541 _ = self.tick_timeout.tick() => {
542 },
546 command = self.service_rx.select_next_some() =>
547 self.process_service_command(command),
548 notification_event = self.notification_service.next_event() => match notification_event {
549 Some(event) => self.process_notification_event(event),
550 None => {
551 error!(
552 target: LOG_TARGET,
553 "Terminating `SyncingEngine` because `NotificationService` has terminated.",
554 );
555
556 return;
557 }
558 },
559 response_event = self.pending_responses.select_next_some() =>
560 self.process_response_event(response_event),
561 validation_result = self.block_announce_validator.select_next_some() =>
562 self.process_block_announce_validation_result(validation_result),
563 }
564
565 self.is_major_syncing.store(self.strategy.is_major_syncing(), Ordering::Relaxed);
567
568 if let Err(e) = self.process_strategy_actions() {
570 error!(
571 target: LOG_TARGET,
572 "Terminating `SyncingEngine` due to fatal error: {e:?}.",
573 );
574 return;
575 }
576 }
577 }
578
579 fn process_strategy_actions(&mut self) -> Result<(), ClientError> {
580 for action in self.strategy.actions(&self.network_service)? {
581 match action {
582 SyncingAction::StartRequest { peer_id, key, request, remove_obsolete } => {
583 if !self.peers.contains_key(&peer_id) {
584 trace!(
585 target: LOG_TARGET,
586 "Cannot start request with strategy key {key:?} to unknown peer \
587 {peer_id}",
588 );
589 debug_assert!(false);
590 continue;
591 }
592 if remove_obsolete {
593 if self.pending_responses.remove(peer_id, key) {
594 warn!(
595 target: LOG_TARGET,
596 "Processed `SyncingAction::StartRequest` to {peer_id} with \
597 strategy key {key:?}. Stale response removed!",
598 )
599 } else {
600 trace!(
601 target: LOG_TARGET,
602 "Processed `SyncingAction::StartRequest` to {peer_id} with \
603 strategy key {key:?}.",
604 )
605 }
606 }
607
608 self.pending_responses.insert(peer_id, key, request);
609 },
610 SyncingAction::CancelRequest { peer_id, key } => {
611 let removed = self.pending_responses.remove(peer_id, key);
612
613 trace!(
614 target: LOG_TARGET,
615 "Processed `SyncingAction::CancelRequest`, response removed: {removed}.",
616 );
617 },
618 SyncingAction::DropPeer(BadPeer(peer_id, rep)) => {
619 self.pending_responses.remove_all(&peer_id);
620 self.network_service
621 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
622 self.network_service.report_peer(peer_id, rep);
623
624 trace!(target: LOG_TARGET, "{peer_id:?} dropped: {rep:?}.");
625 },
626 SyncingAction::ImportBlocks { origin, blocks } => {
627 let count = blocks.len();
628 self.import_blocks(origin, blocks);
629
630 trace!(
631 target: LOG_TARGET,
632 "Processed `ChainSyncAction::ImportBlocks` with {count} blocks.",
633 );
634 },
635 SyncingAction::ImportJustifications { peer_id, hash, number, justifications } => {
636 self.import_justifications(peer_id, hash, number, justifications);
637
638 trace!(
639 target: LOG_TARGET,
640 "Processed `ChainSyncAction::ImportJustifications` from peer {} for block {} ({}).",
641 peer_id,
642 hash,
643 number,
644 )
645 },
646 SyncingAction::Finished => {},
648 }
649 }
650
651 Ok(())
652 }
653
654 fn process_service_command(&mut self, command: ToServiceCommand<B>) {
655 match command {
656 ToServiceCommand::SetSyncForkRequest(peers, hash, number) => {
657 self.strategy.set_sync_fork_request(peers, &hash, number);
658 },
659 ToServiceCommand::EventStream(tx) => {
660 for peer_id in self.peers.keys() {
662 let _ = tx.unbounded_send(SyncEvent::PeerConnected(*peer_id));
663 }
664 self.event_streams.push(tx);
665 },
666 ToServiceCommand::RequestJustification(hash, number) =>
667 self.strategy.request_justification(&hash, number),
668 ToServiceCommand::ClearJustificationRequests =>
669 self.strategy.clear_justification_requests(),
670 ToServiceCommand::BlocksProcessed(imported, count, results) => {
671 self.strategy.on_blocks_processed(imported, count, results);
672 },
673 ToServiceCommand::JustificationImported(peer_id, hash, number, import_result) => {
674 let success =
675 matches!(import_result, sc_consensus::JustificationImportResult::Success);
676 self.strategy.on_justification_import(hash, number, success);
677
678 match import_result {
679 sc_consensus::JustificationImportResult::OutdatedJustification => {
680 log::info!(
681 target: LOG_TARGET,
682 "๐ Outdated justification provided by {peer_id} for #{hash}",
683 );
684 },
685 sc_consensus::JustificationImportResult::Failure => {
686 log::info!(
687 target: LOG_TARGET,
688 "๐ Invalid justification provided by {peer_id} for #{hash}",
689 );
690 self.network_service
691 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
692 self.network_service.report_peer(
693 peer_id,
694 ReputationChange::new_fatal("Invalid justification"),
695 );
696 },
697 sc_consensus::JustificationImportResult::Success => {
698 log::debug!(
699 target: LOG_TARGET,
700 "Justification for block #{hash} ({number}) imported from {peer_id} successfully",
701 );
702 },
703 }
704 },
705 ToServiceCommand::AnnounceBlock(hash, data) => self.announce_block(hash, data),
706 ToServiceCommand::NewBestBlockImported(hash, number) => {
707 log::debug!(target: LOG_TARGET, "New best block imported {:?}/#{}", hash, number);
708
709 self.strategy.update_chain_info(&hash, number);
710 let _ = self.notification_service.try_set_handshake(
711 BlockAnnouncesHandshake::<B>::build(
712 self.roles,
713 number,
714 hash,
715 self.genesis_hash,
716 )
717 .encode(),
718 );
719 },
720 ToServiceCommand::Status(tx) => {
721 let _ = tx.send(self.strategy.status());
722 },
723 ToServiceCommand::NumActivePeers(tx) => {
724 let _ = tx.send(self.num_active_peers());
725 },
726 ToServiceCommand::NumDownloadedBlocks(tx) => {
727 let _ = tx.send(self.strategy.num_downloaded_blocks());
728 },
729 ToServiceCommand::NumSyncRequests(tx) => {
730 let _ = tx.send(self.strategy.num_sync_requests());
731 },
732 ToServiceCommand::PeersInfo(tx) => {
733 let peers_info =
734 self.peers.iter().map(|(peer_id, peer)| (*peer_id, peer.info)).collect();
735 let _ = tx.send(peers_info);
736 },
737 ToServiceCommand::OnBlockFinalized(hash, header) =>
738 self.strategy.on_block_finalized(&hash, *header.number()),
739 }
740 }
741
742 fn process_notification_event(&mut self, event: NotificationEvent) {
743 match event {
744 NotificationEvent::ValidateInboundSubstream { peer, handshake, result_tx } => {
745 let validation_result = self
746 .validate_connection(&peer, handshake, Direction::Inbound)
747 .map_or(ValidationResult::Reject, |_| ValidationResult::Accept);
748
749 let _ = result_tx.send(validation_result);
750 },
751 NotificationEvent::NotificationStreamOpened { peer, handshake, direction, .. } => {
752 log::debug!(
753 target: LOG_TARGET,
754 "Substream opened for {peer}, handshake {handshake:?}"
755 );
756
757 match self.validate_connection(&peer, handshake, direction) {
758 Ok(handshake) => {
759 if self.on_sync_peer_connected(peer, &handshake, direction).is_err() {
760 log::debug!(target: LOG_TARGET, "Failed to register peer {peer}");
761 self.network_service
762 .disconnect_peer(peer, self.block_announce_protocol_name.clone());
763 }
764 },
765 Err(wrong_genesis) => {
766 log::debug!(target: LOG_TARGET, "`SyncingEngine` rejected {peer}");
767
768 if wrong_genesis {
769 self.peer_store_handle.report_peer(peer, rep::GENESIS_MISMATCH);
770 }
771
772 self.network_service
773 .disconnect_peer(peer, self.block_announce_protocol_name.clone());
774 },
775 }
776 },
777 NotificationEvent::NotificationStreamClosed { peer } => {
778 self.on_sync_peer_disconnected(peer);
779 },
780 NotificationEvent::NotificationReceived { peer, notification } => {
781 if !self.peers.contains_key(&peer) {
782 log::error!(
783 target: LOG_TARGET,
784 "received notification from {peer} who had been earlier refused by `SyncingEngine`",
785 );
786 return;
787 }
788
789 let Ok(announce) = BlockAnnounce::decode(&mut notification.as_ref()) else {
790 log::warn!(target: LOG_TARGET, "failed to decode block announce");
791 return;
792 };
793
794 self.push_block_announce_validation(peer, announce);
795 },
796 }
797 }
798
799 fn on_sync_peer_disconnected(&mut self, peer_id: PeerId) {
803 let Some(info) = self.peers.remove(&peer_id) else {
804 log::debug!(target: LOG_TARGET, "{peer_id} does not exist in `SyncingEngine`");
805 return;
806 };
807 if let Some(metrics) = &self.metrics {
808 metrics.peers.dec();
809 }
810 self.num_connected.fetch_sub(1, Ordering::AcqRel);
811
812 if self.important_peers.contains(&peer_id) {
813 log::warn!(target: LOG_TARGET, "Reserved peer {peer_id} disconnected");
814 } else {
815 log::debug!(target: LOG_TARGET, "{peer_id} disconnected");
816 }
817
818 if !self.default_peers_set_no_slot_connected_peers.remove(&peer_id) &&
819 info.inbound &&
820 info.info.roles.is_full()
821 {
822 match self.num_in_peers.checked_sub(1) {
823 Some(value) => {
824 self.num_in_peers = value;
825 },
826 None => {
827 log::error!(
828 target: LOG_TARGET,
829 "trying to disconnect an inbound node which is not counted as inbound"
830 );
831 debug_assert!(false);
832 },
833 }
834 }
835
836 self.strategy.remove_peer(&peer_id);
837 self.pending_responses.remove_all(&peer_id);
838 self.event_streams
839 .retain(|stream| stream.unbounded_send(SyncEvent::PeerDisconnected(peer_id)).is_ok());
840 }
841
842 fn validate_handshake(
844 &mut self,
845 peer_id: &PeerId,
846 handshake: Vec<u8>,
847 ) -> Result<BlockAnnouncesHandshake<B>, bool> {
848 log::trace!(target: LOG_TARGET, "Validate handshake for {peer_id}");
849
850 let handshake = <BlockAnnouncesHandshake<B> as DecodeAll>::decode_all(&mut &handshake[..])
851 .map_err(|error| {
852 log::debug!(target: LOG_TARGET, "Failed to decode handshake for {peer_id}: {error:?}");
853 false
854 })?;
855
856 if handshake.genesis_hash != self.genesis_hash {
857 if self.important_peers.contains(&peer_id) {
858 log::error!(
859 target: LOG_TARGET,
860 "Reserved peer id `{peer_id}` is on a different chain (our genesis: {} theirs: {})",
861 self.genesis_hash,
862 handshake.genesis_hash,
863 );
864 } else if self.boot_node_ids.contains(&peer_id) {
865 log::error!(
866 target: LOG_TARGET,
867 "Bootnode with peer id `{peer_id}` is on a different chain (our genesis: {} theirs: {})",
868 self.genesis_hash,
869 handshake.genesis_hash,
870 );
871 } else {
872 log::debug!(
873 target: LOG_TARGET,
874 "Peer is on different chain (our genesis: {} theirs: {})",
875 self.genesis_hash,
876 handshake.genesis_hash
877 );
878 }
879
880 return Err(true);
881 }
882
883 Ok(handshake)
884 }
885
886 fn validate_connection(
900 &mut self,
901 peer_id: &PeerId,
902 handshake: Vec<u8>,
903 direction: Direction,
904 ) -> Result<BlockAnnouncesHandshake<B>, bool> {
905 log::trace!(target: LOG_TARGET, "New peer {peer_id} {handshake:?}");
906
907 let handshake = self.validate_handshake(peer_id, handshake)?;
908
909 if self.peers.contains_key(&peer_id) {
910 log::error!(
911 target: LOG_TARGET,
912 "Called `validate_connection()` with already connected peer {peer_id}",
913 );
914 debug_assert!(false);
915 return Err(false);
916 }
917
918 let no_slot_peer = self.default_peers_set_no_slot_peers.contains(&peer_id);
919 let this_peer_reserved_slot: usize = if no_slot_peer { 1 } else { 0 };
920
921 if handshake.roles.is_full() &&
922 self.strategy.num_peers() >=
923 self.default_peers_set_num_full +
924 self.default_peers_set_no_slot_connected_peers.len() +
925 this_peer_reserved_slot
926 {
927 log::debug!(target: LOG_TARGET, "Too many full nodes, rejecting {peer_id}");
928 return Err(false);
929 }
930
931 if !no_slot_peer &&
933 handshake.roles.is_full() &&
934 direction.is_inbound() &&
935 self.num_in_peers == self.max_in_peers
936 {
937 log::debug!(target: LOG_TARGET, "All inbound slots have been consumed, rejecting {peer_id}");
938 return Err(false);
939 }
940
941 if handshake.roles.is_light() &&
946 (self.peers.len() - self.strategy.num_peers()) >= self.default_peers_set_num_light
947 {
948 log::debug!(target: LOG_TARGET, "Too many light nodes, rejecting {peer_id}");
949 return Err(false);
950 }
951
952 Ok(handshake)
953 }
954
955 fn on_sync_peer_connected(
961 &mut self,
962 peer_id: PeerId,
963 status: &BlockAnnouncesHandshake<B>,
964 direction: Direction,
965 ) -> Result<(), ()> {
966 log::trace!(target: LOG_TARGET, "New peer {peer_id} {status:?}");
967
968 let peer = Peer {
969 info: ExtendedPeerInfo {
970 roles: status.roles,
971 best_hash: status.best_hash,
972 best_number: status.best_number,
973 },
974 known_blocks: LruHashSet::new(
975 NonZeroUsize::new(MAX_KNOWN_BLOCKS).expect("Constant is nonzero"),
976 ),
977 inbound: direction.is_inbound(),
978 };
979
980 if status.roles.is_full() {
982 self.strategy.add_peer(peer_id, peer.info.best_hash, peer.info.best_number);
983 }
984
985 log::debug!(target: LOG_TARGET, "Connected {peer_id}");
986
987 if self.peers.insert(peer_id, peer).is_none() {
988 if let Some(metrics) = &self.metrics {
989 metrics.peers.inc();
990 }
991 self.num_connected.fetch_add(1, Ordering::AcqRel);
992 }
993 self.peer_store_handle.set_peer_role(&peer_id, status.roles.into());
994
995 if self.default_peers_set_no_slot_peers.contains(&peer_id) {
996 self.default_peers_set_no_slot_connected_peers.insert(peer_id);
997 } else if direction.is_inbound() && status.roles.is_full() {
998 self.num_in_peers += 1;
999 }
1000
1001 self.event_streams
1002 .retain(|stream| stream.unbounded_send(SyncEvent::PeerConnected(peer_id)).is_ok());
1003
1004 Ok(())
1005 }
1006
1007 fn process_response_event(&mut self, response_event: ResponseEvent) {
1008 let ResponseEvent { peer_id, key, response: response_result } = response_event;
1009
1010 match response_result {
1011 Ok(Ok((response, protocol_name))) => {
1012 self.strategy.on_generic_response(&peer_id, key, protocol_name, response);
1013 },
1014 Ok(Err(e)) => {
1015 debug!(target: LOG_TARGET, "Request to peer {peer_id:?} failed: {e:?}.");
1016
1017 match e {
1018 RequestFailure::Network(OutboundFailure::Timeout) => {
1019 self.network_service.report_peer(peer_id, rep::TIMEOUT);
1020 self.network_service
1021 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
1022 },
1023 RequestFailure::Network(OutboundFailure::UnsupportedProtocols) => {
1024 self.network_service.report_peer(peer_id, rep::BAD_PROTOCOL);
1025 self.network_service
1026 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
1027 },
1028 RequestFailure::Network(OutboundFailure::DialFailure) => {
1029 self.network_service
1030 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
1031 },
1032 RequestFailure::Refused => {
1033 self.network_service.report_peer(peer_id, rep::REFUSED);
1034 self.network_service
1035 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
1036 },
1037 RequestFailure::Network(OutboundFailure::ConnectionClosed) |
1038 RequestFailure::NotConnected => {
1039 self.network_service
1040 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
1041 },
1042 RequestFailure::UnknownProtocol => {
1043 debug_assert!(false, "Block request protocol should always be known.");
1044 },
1045 RequestFailure::Obsolete => {
1046 debug_assert!(
1047 false,
1048 "Can not receive `RequestFailure::Obsolete` after dropping the \
1049 response receiver.",
1050 );
1051 },
1052 RequestFailure::Network(OutboundFailure::Io(_)) => {
1053 self.network_service.report_peer(peer_id, rep::IO);
1054 self.network_service
1055 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
1056 },
1057 }
1058 },
1059 Err(oneshot::Canceled) => {
1060 trace!(
1061 target: LOG_TARGET,
1062 "Request to peer {peer_id:?} failed due to oneshot being canceled.",
1063 );
1064 self.network_service
1065 .disconnect_peer(peer_id, self.block_announce_protocol_name.clone());
1066 },
1067 }
1068 }
1069
1070 fn num_active_peers(&self) -> usize {
1072 self.pending_responses.len()
1073 }
1074
1075 fn get_block_announce_proto_config<N: NetworkBackend<B, <B as BlockT>::Hash>>(
1077 protocol_id: ProtocolId,
1078 fork_id: Option<&str>,
1079 roles: Roles,
1080 best_number: NumberFor<B>,
1081 best_hash: B::Hash,
1082 genesis_hash: B::Hash,
1083 set_config: &SetConfig,
1084 metrics: NotificationMetrics,
1085 peer_store_handle: Arc<dyn PeerStoreProvider>,
1086 ) -> (N::NotificationProtocolConfig, Box<dyn NotificationService>) {
1087 let block_announces_protocol = {
1088 let genesis_hash = genesis_hash.as_ref();
1089 if let Some(fork_id) = fork_id {
1090 format!(
1091 "/{}/{}/block-announces/1",
1092 array_bytes::bytes2hex("", genesis_hash),
1093 fork_id
1094 )
1095 } else {
1096 format!("/{}/block-announces/1", array_bytes::bytes2hex("", genesis_hash))
1097 }
1098 };
1099
1100 N::notification_config(
1101 block_announces_protocol.into(),
1102 iter::once(format!("/{}/block-announces/1", protocol_id.as_ref()).into()).collect(),
1103 MAX_BLOCK_ANNOUNCE_SIZE,
1104 Some(NotificationHandshake::new(BlockAnnouncesHandshake::<B>::build(
1105 roles,
1106 best_number,
1107 best_hash,
1108 genesis_hash,
1109 ))),
1110 set_config.clone(),
1111 metrics,
1112 peer_store_handle,
1113 )
1114 }
1115
1116 fn import_blocks(&mut self, origin: BlockOrigin, blocks: Vec<IncomingBlock<B>>) {
1118 if let Some(metrics) = &self.metrics {
1119 metrics.import_queue_blocks_submitted.inc();
1120 }
1121
1122 self.import_queue.import_blocks(origin, blocks);
1123 }
1124
1125 fn import_justifications(
1127 &mut self,
1128 peer_id: PeerId,
1129 hash: B::Hash,
1130 number: NumberFor<B>,
1131 justifications: Justifications,
1132 ) {
1133 if let Some(metrics) = &self.metrics {
1134 metrics.import_queue_justifications_submitted.inc();
1135 }
1136
1137 self.import_queue.import_justifications(peer_id, hash, number, justifications);
1138 }
1139}