1use crate::LOG_TARGET;
20use log::trace;
21use prometheus_endpoint::{register, Counter, PrometheusError, Registry, U64};
22use sc_network_common::sync::message;
23use sc_network_types::PeerId;
24use sp_arithmetic::traits::Saturating;
25use sp_runtime::traits::{Block as BlockT, NumberFor, One};
26use std::{
27 cmp,
28 collections::{BTreeMap, HashMap},
29 ops::Range,
30};
31
32#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct BlockData<B: BlockT> {
35 pub block: message::BlockData<B>,
37 pub origin: Option<PeerId>,
39}
40
41#[derive(Debug)]
42enum BlockRangeState<B: BlockT> {
43 Downloading { len: NumberFor<B>, downloading: u32 },
44 Complete(Vec<BlockData<B>>),
45 Queued { len: NumberFor<B> },
46}
47
48impl<B: BlockT> BlockRangeState<B> {
49 pub fn len(&self) -> NumberFor<B> {
50 match *self {
51 Self::Downloading { len, .. } => len,
52 Self::Complete(ref blocks) => (blocks.len() as u32).into(),
53 Self::Queued { len } => len,
54 }
55 }
56}
57
58#[derive(Clone)]
59pub(crate) struct Metrics {
60 stale_download_reservations: Counter<U64>,
61}
62
63impl Metrics {
64 pub(crate) fn register(registry: &Registry) -> Result<Self, PrometheusError> {
65 Ok(Self {
66 stale_download_reservations: register(
67 Counter::new(
68 "substrate_sync_stale_download_reservations_total",
69 "Number of stale block download reservations released before scheduling a new range",
70 )?,
71 registry,
72 )?,
73 })
74 }
75}
76
77#[derive(Default)]
79pub struct BlockCollection<B: BlockT> {
80 blocks: BTreeMap<NumberFor<B>, BlockRangeState<B>>,
82 peer_requests: HashMap<PeerId, NumberFor<B>>,
83 queued_blocks: HashMap<B::Hash, (NumberFor<B>, NumberFor<B>)>,
86 metrics: Option<Metrics>,
88}
89
90impl<B: BlockT> BlockCollection<B> {
91 pub fn new() -> Self {
93 Self::with_metrics(None)
94 }
95
96 pub(crate) fn with_metrics(metrics: Option<Metrics>) -> Self {
97 Self {
98 blocks: BTreeMap::new(),
99 peer_requests: HashMap::new(),
100 queued_blocks: HashMap::new(),
101 metrics,
102 }
103 }
104
105 pub fn clear(&mut self) {
107 self.blocks.clear();
108 self.peer_requests.clear();
109 }
110
111 pub fn insert(&mut self, start: NumberFor<B>, blocks: Vec<message::BlockData<B>>, who: PeerId) {
113 if blocks.is_empty() {
114 return;
115 }
116
117 match self.blocks.get(&start) {
118 Some(&BlockRangeState::Downloading { .. }) => {
119 trace!(target: LOG_TARGET, "Inserting block data still marked as being downloaded: {}", start);
120 },
121 Some(BlockRangeState::Complete(existing)) if existing.len() >= blocks.len() => {
122 trace!(target: LOG_TARGET, "Ignored block data already downloaded: {}", start);
123 return;
124 },
125 _ => (),
126 }
127
128 self.blocks.insert(
129 start,
130 BlockRangeState::Complete(
131 blocks.into_iter().map(|b| BlockData { origin: Some(who), block: b }).collect(),
132 ),
133 );
134 }
135
136 pub fn needed_blocks(
139 &mut self,
140 who: PeerId,
141 count: u32,
142 peer_best: NumberFor<B>,
143 common: NumberFor<B>,
144 max_parallel: u32,
145 max_ahead: u32,
146 ) -> Option<Range<NumberFor<B>>> {
147 if self.peer_requests.contains_key(&who) {
149 log::debug!(target: LOG_TARGET, "Releasing stale block download reservation for {who}");
150 if let Some(metrics) = &self.metrics {
151 metrics.stale_download_reservations.inc();
152 }
153 self.clear_peer_download(&who);
154 }
155 if peer_best <= common {
156 return None;
158 }
159 let first_different = common + <NumberFor<B>>::one();
161 let count = (count as u32).into();
162 let (mut range, downloading) = {
163 let mut downloading_iter = self.blocks.iter().peekable();
165 let mut prev: Option<(&NumberFor<B>, &BlockRangeState<B>)> = None;
166 loop {
167 let next = downloading_iter.next();
168 break match (prev, next) {
169 (Some((start, &BlockRangeState::Downloading { ref len, downloading })), _)
174 if downloading < max_parallel && *start >= first_different =>
175 {
176 (*start..*start + *len, downloading)
177 },
178 (Some((start, r)), Some((next_start, _)))
181 if *start + r.len() < *next_start &&
182 *start + r.len() >= first_different =>
183 {
184 (*start + r.len()..cmp::min(*next_start, *start + r.len() + count), 0)
185 },
186 (Some((start, r)), None) if *start + r.len() >= first_different => {
189 (*start + r.len()..*start + r.len() + count, 0)
190 },
191 (None, None) => (first_different..first_different + count, 0),
194 (None, Some((start, _))) if *start > first_different => {
196 (first_different..cmp::min(first_different + count, *start), 0)
197 },
198 _ => {
200 prev = next;
201 continue;
202 },
203 };
204 }
205 };
206 if range.start > peer_best {
208 trace!(target: LOG_TARGET, "Out of range for peer {} ({} vs {})", who, range.start, peer_best);
209 return None;
210 }
211
212 range.end = cmp::min(peer_best.saturating_add(One::one()), range.end);
213
214 if self
215 .blocks
216 .iter()
217 .next()
218 .map_or(false, |(n, _)| range.start > *n + max_ahead.into())
219 {
220 trace!(target: LOG_TARGET, "Too far ahead for peer {} ({})", who, range.start);
221 return None;
222 }
223
224 if range.end <= range.start {
225 debug_assert!(
226 false,
227 "Empty range {:?}, count={}, peer_best={}, common={}, blocks={:?}",
228 range, count, peer_best, common, self.blocks
229 );
230 trace!(
231 target: LOG_TARGET,
232 "Empty range for peer {who}: {range:?}, count={count}, peer_best={peer_best}, common={common}",
233 );
234 return None;
235 }
236
237 self.peer_requests.insert(who, range.start);
238 self.blocks.insert(
239 range.start,
240 BlockRangeState::Downloading {
241 len: range.end - range.start,
242 downloading: downloading + 1,
243 },
244 );
245
246 Some(range)
247 }
248
249 pub fn ready_blocks(&mut self, from: NumberFor<B>) -> Vec<BlockData<B>> {
255 let mut ready = Vec::new();
256
257 let mut prev = from;
258 for (&start, range_data) in &mut self.blocks {
259 if start > prev {
260 break;
261 }
262 let len = match range_data {
263 BlockRangeState::Complete(blocks) => {
264 let len = (blocks.len() as u32).into();
265 prev = start + len;
266 if let Some(BlockData { block, .. }) = blocks.first() {
267 self.queued_blocks
268 .insert(block.hash, (start, start + (blocks.len() as u32).into()));
269 }
270 ready.append(blocks);
272 len
273 },
274 BlockRangeState::Queued { .. } => continue,
275 _ => break,
276 };
277 *range_data = BlockRangeState::Queued { len };
278 }
279 trace!(target: LOG_TARGET, "{} blocks ready for import", ready.len());
280 ready
281 }
282
283 pub fn clear_queued(&mut self, hash: &B::Hash) {
284 if let Some((from, to)) = self.queued_blocks.remove(hash) {
285 let mut block_num = from;
286 while block_num < to {
287 self.blocks.remove(&block_num);
288 block_num += One::one();
289 }
290 trace!(target: LOG_TARGET, "Cleared blocks from {:?} to {:?}", from, to);
291 }
292 }
293
294 pub fn clear_peer_download(&mut self, who: &PeerId) {
295 if let Some(start) = self.peer_requests.remove(who) {
296 let remove = match self.blocks.get_mut(&start) {
297 Some(&mut BlockRangeState::Downloading { ref mut downloading, .. })
298 if *downloading > 1 =>
299 {
300 *downloading -= 1;
301 false
302 },
303 Some(&mut BlockRangeState::Downloading { .. }) => true,
304 _ => false,
305 };
306 if remove {
307 self.blocks.remove(&start);
308 }
309 }
310 }
311}
312
313#[cfg(test)]
314mod test {
315 use super::{BlockCollection, BlockData, BlockRangeState};
316 use sc_network_common::sync::message;
317 use sc_network_types::PeerId;
318 use sp_core::H256;
319 use sp_runtime::testing::{Block as RawBlock, MockCallU64, TestXt};
320
321 type Block = RawBlock<TestXt<MockCallU64, ()>>;
322
323 fn is_empty(bc: &BlockCollection<Block>) -> bool {
324 bc.blocks.is_empty() && bc.peer_requests.is_empty()
325 }
326
327 fn generate_blocks(n: usize) -> Vec<message::BlockData<Block>> {
328 (0..n)
329 .map(|_| message::generic::BlockData {
330 hash: H256::random(),
331 header: None,
332 body: None,
333 indexed_body: None,
334 message_queue: None,
335 receipt: None,
336 justification: None,
337 justifications: None,
338 })
339 .collect()
340 }
341
342 #[test]
343 fn create_clear() {
344 let mut bc = BlockCollection::new();
345 assert!(is_empty(&bc));
346 bc.insert(1, generate_blocks(100), PeerId::random());
347 assert!(!is_empty(&bc));
348 bc.clear();
349 assert!(is_empty(&bc));
350 }
351
352 #[test]
353 fn insert_blocks() {
354 let mut bc = BlockCollection::new();
355 assert!(is_empty(&bc));
356 let peer0 = PeerId::random();
357 let peer1 = PeerId::random();
358 let peer2 = PeerId::random();
359
360 let blocks = generate_blocks(150);
361 assert_eq!(bc.needed_blocks(peer0, 40, 150, 0, 1, 200), Some(1..41));
362 assert_eq!(bc.needed_blocks(peer1, 40, 150, 0, 1, 200), Some(41..81));
363 assert_eq!(bc.needed_blocks(peer2, 40, 150, 0, 1, 200), Some(81..121));
364
365 bc.clear_peer_download(&peer1);
366 bc.insert(41, blocks[41..81].to_vec(), peer1);
367 assert_eq!(bc.ready_blocks(1), vec![]);
368 assert_eq!(bc.needed_blocks(peer1, 40, 150, 0, 1, 200), Some(121..151));
369 bc.clear_peer_download(&peer0);
370 bc.insert(1, blocks[1..11].to_vec(), peer0);
371
372 assert_eq!(bc.needed_blocks(peer0, 40, 150, 0, 1, 200), Some(11..41));
373 assert_eq!(
374 bc.ready_blocks(1),
375 blocks[1..11]
376 .iter()
377 .map(|b| BlockData { block: b.clone(), origin: Some(peer0) })
378 .collect::<Vec<_>>()
379 );
380
381 bc.clear_peer_download(&peer0);
382 bc.insert(11, blocks[11..41].to_vec(), peer0);
383
384 let ready = bc.ready_blocks(12);
385 assert_eq!(
386 ready[..30],
387 blocks[11..41]
388 .iter()
389 .map(|b| BlockData { block: b.clone(), origin: Some(peer0) })
390 .collect::<Vec<_>>()[..]
391 );
392 assert_eq!(
393 ready[30..],
394 blocks[41..81]
395 .iter()
396 .map(|b| BlockData { block: b.clone(), origin: Some(peer1) })
397 .collect::<Vec<_>>()[..]
398 );
399
400 bc.clear_peer_download(&peer2);
401 assert_eq!(bc.needed_blocks(peer2, 40, 150, 80, 1, 200), Some(81..121));
402 bc.clear_peer_download(&peer2);
403 bc.insert(81, blocks[81..121].to_vec(), peer2);
404 bc.clear_peer_download(&peer1);
405 bc.insert(121, blocks[121..150].to_vec(), peer1);
406
407 assert_eq!(bc.ready_blocks(80), vec![]);
408 let ready = bc.ready_blocks(81);
409 assert_eq!(
410 ready[..40],
411 blocks[81..121]
412 .iter()
413 .map(|b| BlockData { block: b.clone(), origin: Some(peer2) })
414 .collect::<Vec<_>>()[..]
415 );
416 assert_eq!(
417 ready[40..],
418 blocks[121..150]
419 .iter()
420 .map(|b| BlockData { block: b.clone(), origin: Some(peer1) })
421 .collect::<Vec<_>>()[..]
422 );
423 }
424
425 #[test]
426 fn large_gap() {
427 let mut bc: BlockCollection<Block> = BlockCollection::new();
428 bc.blocks.insert(100, BlockRangeState::Downloading { len: 128, downloading: 1 });
429 let blocks = generate_blocks(10)
430 .into_iter()
431 .map(|b| BlockData { block: b, origin: None })
432 .collect();
433 bc.blocks.insert(114305, BlockRangeState::Complete(blocks));
434
435 let peer0 = PeerId::random();
438 let peer1 = PeerId::random();
439 let peer2 = PeerId::random();
440 assert_eq!(bc.needed_blocks(peer0, 128, 10000, 0, 1, 200), Some(1..100));
441 assert_eq!(bc.needed_blocks(peer1, 128, 10000, 0, 1, 200), None); assert_eq!(
443 bc.needed_blocks(peer2, 128, 10000, 0, 1, 200000),
444 Some(100 + 128..100 + 128 + 128)
445 );
446 }
447
448 #[test]
449 fn no_duplicate_requests_on_fork() {
450 let mut bc = BlockCollection::new();
451 assert!(is_empty(&bc));
452 let peer = PeerId::random();
453
454 let blocks = generate_blocks(10);
455
456 assert_eq!(bc.needed_blocks(peer, 5, 50, 39, 0, 200), Some(40..45));
458
459 bc.clear_peer_download(&peer);
461 bc.insert(40, blocks[..5].to_vec(), peer);
462
463 let ready = bc.ready_blocks(48);
465 assert_eq!(
466 ready,
467 blocks[..5]
468 .iter()
469 .map(|b| BlockData { block: b.clone(), origin: Some(peer) })
470 .collect::<Vec<_>>()
471 );
472
473 assert_eq!(bc.needed_blocks(peer, 5, 50, 39, 0, 200), Some(45..50));
474 }
475
476 #[test]
477 fn clear_queued_subsequent_ranges() {
478 let mut bc = BlockCollection::new();
479 assert!(is_empty(&bc));
480 let peer1 = PeerId::random();
483 let peer2 = PeerId::random();
484
485 let blocks = generate_blocks(10);
486
487 assert_eq!(bc.needed_blocks(peer1, 5, 50, 39, 0, 200), Some(40..45));
489 assert_eq!(bc.needed_blocks(peer2, 5, 50, 39, 0, 200), Some(45..50));
490
491 bc.clear_peer_download(&peer1);
493 bc.clear_peer_download(&peer2);
494 bc.insert(40, blocks.to_vec(), peer1);
495
496 let ready = bc.ready_blocks(1000);
498 assert_eq!(
499 ready,
500 blocks
501 .iter()
502 .map(|b| BlockData { block: b.clone(), origin: Some(peer1) })
503 .collect::<Vec<_>>()
504 );
505
506 bc.clear_queued(&blocks[0].hash);
507 assert!(bc.blocks.is_empty());
508 assert!(bc.queued_blocks.is_empty());
509 }
510
511 #[test]
512 fn downloaded_range_is_requested_from_max_parallel_peers() {
513 let mut bc = BlockCollection::new();
514 assert!(is_empty(&bc));
515
516 let count = 5;
517 let max_parallel = 2;
519 let max_ahead = 200;
520
521 let peer1 = PeerId::random();
522 let peer2 = PeerId::random();
523 let peer3 = PeerId::random();
524
525 let best = 100;
527 let common = 10;
528
529 assert_eq!(
530 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
531 Some(11..16)
532 );
533 assert_eq!(
534 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
535 Some(11..16)
536 );
537 assert_eq!(
538 bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
539 Some(16..21)
540 );
541 }
542 #[test]
543 fn downloaded_range_not_requested_from_peers_with_higher_common_number() {
544 let mut bc = BlockCollection::new();
551 assert!(is_empty(&bc));
552
553 let count = 5;
554 let max_parallel = 2;
555 let max_ahead = 200;
556
557 let peer1 = PeerId::random();
558 let peer1_best = 20;
559 let peer1_common = 10;
560
561 let peer2 = PeerId::random();
563 let peer2_best = 20;
564 let peer2_common = 11; assert_eq!(
567 bc.needed_blocks(peer1, count, peer1_best, peer1_common, max_parallel, max_ahead),
568 Some(11..16),
569 );
570 assert_eq!(
571 bc.needed_blocks(peer2, count, peer2_best, peer2_common, max_parallel, max_ahead),
572 Some(16..21),
573 );
574 }
575
576 #[test]
577 fn requesting_a_new_range_releases_a_peers_stale_range() {
578 let metrics = super::Metrics::register(&prometheus_endpoint::Registry::new()).unwrap();
583 let mut bc = BlockCollection::with_metrics(Some(metrics.clone()));
584 assert!(is_empty(&bc));
585
586 let count = 128;
587 let best = 10_000;
588 let max_parallel = 1;
589 let max_ahead = 2048;
590
591 let peer = PeerId::random();
592
593 let first = bc.needed_blocks(peer, count, best, 0, max_parallel, max_ahead).unwrap();
595 assert_eq!(bc.peer_requests.get(&peer), Some(&first.start));
596 assert!(matches!(bc.blocks.get(&first.start), Some(BlockRangeState::Downloading { .. }),));
597 assert_eq!(metrics.stale_download_reservations.get(), 0);
598
599 let second = bc.needed_blocks(peer, count, best, 200, max_parallel, max_ahead).unwrap();
603 assert_eq!(metrics.stale_download_reservations.get(), 1);
604 assert_ne!(first.start, second.start);
605 assert_eq!(bc.peer_requests.get(&peer), Some(&second.start));
606
607 let downloading = bc
608 .blocks
609 .iter()
610 .filter(|(_, state)| matches!(state, BlockRangeState::Downloading { .. }))
611 .map(|(n, _)| *n)
612 .collect::<Vec<_>>();
613 assert_eq!(
614 downloading,
615 vec![second.start],
616 "stale range must be released, leaving no orphaned Downloading marker",
617 );
618
619 bc.clear_peer_download(&peer);
622 assert!(bc.needed_blocks(peer, count, best, 400, max_parallel, max_ahead).is_some());
623 bc.clear();
624 assert!(bc.needed_blocks(peer, count, best, 600, max_parallel, max_ahead).is_some());
625 assert_eq!(metrics.stale_download_reservations.get(), 1);
626 }
627
628 #[test]
629 fn gap_above_common_number_requested() {
630 let mut bc = BlockCollection::new();
631 assert!(is_empty(&bc));
632
633 let count = 5;
634 let best = 30;
635 let max_parallel = 1;
638 let max_ahead = 200;
639
640 let peer1 = PeerId::random();
641 let peer2 = PeerId::random();
642 let peer3 = PeerId::random();
643
644 let common = 10;
645 assert_eq!(
646 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
647 Some(11..16),
648 );
649 assert_eq!(
650 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
651 Some(16..21),
652 );
653 assert_eq!(
654 bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
655 Some(21..26),
656 );
657
658 bc.clear_peer_download(&peer2);
661
662 assert_eq!(
664 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
665 Some(16..21),
666 );
667 }
668
669 #[test]
670 fn gap_below_common_number_not_requested() {
671 let mut bc = BlockCollection::new();
672 assert!(is_empty(&bc));
673
674 let count = 5;
675 let best = 30;
676 let max_parallel = 1;
679 let max_ahead = 200;
680
681 let peer1 = PeerId::random();
682 let peer2 = PeerId::random();
683 let peer3 = PeerId::random();
684
685 let common = 10;
686 assert_eq!(
687 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
688 Some(11..16),
689 );
690 assert_eq!(
691 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
692 Some(16..21),
693 );
694 assert_eq!(
695 bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
696 Some(21..26),
697 );
698
699 bc.clear_peer_download(&peer2);
702
703 let common = 23;
705 assert_eq!(
706 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
707 Some(26..31), );
709 }
710
711 #[test]
712 fn range_at_the_end_above_common_number_requested() {
713 let mut bc = BlockCollection::new();
714 assert!(is_empty(&bc));
715
716 let count = 5;
717 let best = 30;
718 let max_parallel = 1;
719 let max_ahead = 200;
720
721 let peer1 = PeerId::random();
722 let peer2 = PeerId::random();
723
724 let common = 10;
725 assert_eq!(
726 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
727 Some(11..16),
728 );
729 assert_eq!(
730 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
731 Some(16..21),
732 );
733 }
734
735 #[test]
736 fn range_at_the_end_below_common_number_not_requested() {
737 let mut bc = BlockCollection::new();
738 assert!(is_empty(&bc));
739
740 let count = 5;
741 let best = 30;
742 let max_parallel = 1;
743 let max_ahead = 200;
744
745 let peer1 = PeerId::random();
746 let peer2 = PeerId::random();
747
748 let common = 10;
749 assert_eq!(
750 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
751 Some(11..16),
752 );
753
754 let common = 20;
755 assert_eq!(
756 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
757 Some(21..26), );
759 }
760}