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 message_queue: None,
334 receipt: None,
335 justification: None,
336 justifications: None,
337 })
338 .collect()
339 }
340
341 #[test]
342 fn create_clear() {
343 let mut bc = BlockCollection::new();
344 assert!(is_empty(&bc));
345 bc.insert(1, generate_blocks(100), PeerId::random());
346 assert!(!is_empty(&bc));
347 bc.clear();
348 assert!(is_empty(&bc));
349 }
350
351 #[test]
352 fn insert_blocks() {
353 let mut bc = BlockCollection::new();
354 assert!(is_empty(&bc));
355 let peer0 = PeerId::random();
356 let peer1 = PeerId::random();
357 let peer2 = PeerId::random();
358
359 let blocks = generate_blocks(150);
360 assert_eq!(bc.needed_blocks(peer0, 40, 150, 0, 1, 200), Some(1..41));
361 assert_eq!(bc.needed_blocks(peer1, 40, 150, 0, 1, 200), Some(41..81));
362 assert_eq!(bc.needed_blocks(peer2, 40, 150, 0, 1, 200), Some(81..121));
363
364 bc.clear_peer_download(&peer1);
365 bc.insert(41, blocks[41..81].to_vec(), peer1);
366 assert_eq!(bc.ready_blocks(1), vec![]);
367 assert_eq!(bc.needed_blocks(peer1, 40, 150, 0, 1, 200), Some(121..151));
368 bc.clear_peer_download(&peer0);
369 bc.insert(1, blocks[1..11].to_vec(), peer0);
370
371 assert_eq!(bc.needed_blocks(peer0, 40, 150, 0, 1, 200), Some(11..41));
372 assert_eq!(
373 bc.ready_blocks(1),
374 blocks[1..11]
375 .iter()
376 .map(|b| BlockData { block: b.clone(), origin: Some(peer0) })
377 .collect::<Vec<_>>()
378 );
379
380 bc.clear_peer_download(&peer0);
381 bc.insert(11, blocks[11..41].to_vec(), peer0);
382
383 let ready = bc.ready_blocks(12);
384 assert_eq!(
385 ready[..30],
386 blocks[11..41]
387 .iter()
388 .map(|b| BlockData { block: b.clone(), origin: Some(peer0) })
389 .collect::<Vec<_>>()[..]
390 );
391 assert_eq!(
392 ready[30..],
393 blocks[41..81]
394 .iter()
395 .map(|b| BlockData { block: b.clone(), origin: Some(peer1) })
396 .collect::<Vec<_>>()[..]
397 );
398
399 bc.clear_peer_download(&peer2);
400 assert_eq!(bc.needed_blocks(peer2, 40, 150, 80, 1, 200), Some(81..121));
401 bc.clear_peer_download(&peer2);
402 bc.insert(81, blocks[81..121].to_vec(), peer2);
403 bc.clear_peer_download(&peer1);
404 bc.insert(121, blocks[121..150].to_vec(), peer1);
405
406 assert_eq!(bc.ready_blocks(80), vec![]);
407 let ready = bc.ready_blocks(81);
408 assert_eq!(
409 ready[..40],
410 blocks[81..121]
411 .iter()
412 .map(|b| BlockData { block: b.clone(), origin: Some(peer2) })
413 .collect::<Vec<_>>()[..]
414 );
415 assert_eq!(
416 ready[40..],
417 blocks[121..150]
418 .iter()
419 .map(|b| BlockData { block: b.clone(), origin: Some(peer1) })
420 .collect::<Vec<_>>()[..]
421 );
422 }
423
424 #[test]
425 fn large_gap() {
426 let mut bc: BlockCollection<Block> = BlockCollection::new();
427 bc.blocks.insert(100, BlockRangeState::Downloading { len: 128, downloading: 1 });
428 let blocks = generate_blocks(10)
429 .into_iter()
430 .map(|b| BlockData { block: b, origin: None })
431 .collect();
432 bc.blocks.insert(114305, BlockRangeState::Complete(blocks));
433
434 let peer0 = PeerId::random();
437 let peer1 = PeerId::random();
438 let peer2 = PeerId::random();
439 assert_eq!(bc.needed_blocks(peer0, 128, 10000, 0, 1, 200), Some(1..100));
440 assert_eq!(bc.needed_blocks(peer1, 128, 10000, 0, 1, 200), None); assert_eq!(
442 bc.needed_blocks(peer2, 128, 10000, 0, 1, 200000),
443 Some(100 + 128..100 + 128 + 128)
444 );
445 }
446
447 #[test]
448 fn no_duplicate_requests_on_fork() {
449 let mut bc = BlockCollection::new();
450 assert!(is_empty(&bc));
451 let peer = PeerId::random();
452
453 let blocks = generate_blocks(10);
454
455 assert_eq!(bc.needed_blocks(peer, 5, 50, 39, 0, 200), Some(40..45));
457
458 bc.clear_peer_download(&peer);
460 bc.insert(40, blocks[..5].to_vec(), peer);
461
462 let ready = bc.ready_blocks(48);
464 assert_eq!(
465 ready,
466 blocks[..5]
467 .iter()
468 .map(|b| BlockData { block: b.clone(), origin: Some(peer) })
469 .collect::<Vec<_>>()
470 );
471
472 assert_eq!(bc.needed_blocks(peer, 5, 50, 39, 0, 200), Some(45..50));
473 }
474
475 #[test]
476 fn clear_queued_subsequent_ranges() {
477 let mut bc = BlockCollection::new();
478 assert!(is_empty(&bc));
479 let peer1 = PeerId::random();
482 let peer2 = PeerId::random();
483
484 let blocks = generate_blocks(10);
485
486 assert_eq!(bc.needed_blocks(peer1, 5, 50, 39, 0, 200), Some(40..45));
488 assert_eq!(bc.needed_blocks(peer2, 5, 50, 39, 0, 200), Some(45..50));
489
490 bc.clear_peer_download(&peer1);
492 bc.clear_peer_download(&peer2);
493 bc.insert(40, blocks.to_vec(), peer1);
494
495 let ready = bc.ready_blocks(1000);
497 assert_eq!(
498 ready,
499 blocks
500 .iter()
501 .map(|b| BlockData { block: b.clone(), origin: Some(peer1) })
502 .collect::<Vec<_>>()
503 );
504
505 bc.clear_queued(&blocks[0].hash);
506 assert!(bc.blocks.is_empty());
507 assert!(bc.queued_blocks.is_empty());
508 }
509
510 #[test]
511 fn downloaded_range_is_requested_from_max_parallel_peers() {
512 let mut bc = BlockCollection::new();
513 assert!(is_empty(&bc));
514
515 let count = 5;
516 let max_parallel = 2;
518 let max_ahead = 200;
519
520 let peer1 = PeerId::random();
521 let peer2 = PeerId::random();
522 let peer3 = PeerId::random();
523
524 let best = 100;
526 let common = 10;
527
528 assert_eq!(
529 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
530 Some(11..16)
531 );
532 assert_eq!(
533 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
534 Some(11..16)
535 );
536 assert_eq!(
537 bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
538 Some(16..21)
539 );
540 }
541 #[test]
542 fn downloaded_range_not_requested_from_peers_with_higher_common_number() {
543 let mut bc = BlockCollection::new();
550 assert!(is_empty(&bc));
551
552 let count = 5;
553 let max_parallel = 2;
554 let max_ahead = 200;
555
556 let peer1 = PeerId::random();
557 let peer1_best = 20;
558 let peer1_common = 10;
559
560 let peer2 = PeerId::random();
562 let peer2_best = 20;
563 let peer2_common = 11; assert_eq!(
566 bc.needed_blocks(peer1, count, peer1_best, peer1_common, max_parallel, max_ahead),
567 Some(11..16),
568 );
569 assert_eq!(
570 bc.needed_blocks(peer2, count, peer2_best, peer2_common, max_parallel, max_ahead),
571 Some(16..21),
572 );
573 }
574
575 #[test]
576 fn requesting_a_new_range_releases_a_peers_stale_range() {
577 let metrics = super::Metrics::register(&prometheus_endpoint::Registry::new()).unwrap();
582 let mut bc = BlockCollection::with_metrics(Some(metrics.clone()));
583 assert!(is_empty(&bc));
584
585 let count = 128;
586 let best = 10_000;
587 let max_parallel = 1;
588 let max_ahead = 2048;
589
590 let peer = PeerId::random();
591
592 let first = bc.needed_blocks(peer, count, best, 0, max_parallel, max_ahead).unwrap();
594 assert_eq!(bc.peer_requests.get(&peer), Some(&first.start));
595 assert!(matches!(bc.blocks.get(&first.start), Some(BlockRangeState::Downloading { .. }),));
596 assert_eq!(metrics.stale_download_reservations.get(), 0);
597
598 let second = bc.needed_blocks(peer, count, best, 200, max_parallel, max_ahead).unwrap();
602 assert_eq!(metrics.stale_download_reservations.get(), 1);
603 assert_ne!(first.start, second.start);
604 assert_eq!(bc.peer_requests.get(&peer), Some(&second.start));
605
606 let downloading = bc
607 .blocks
608 .iter()
609 .filter(|(_, state)| matches!(state, BlockRangeState::Downloading { .. }))
610 .map(|(n, _)| *n)
611 .collect::<Vec<_>>();
612 assert_eq!(
613 downloading,
614 vec![second.start],
615 "stale range must be released, leaving no orphaned Downloading marker",
616 );
617
618 bc.clear_peer_download(&peer);
621 assert!(bc.needed_blocks(peer, count, best, 400, max_parallel, max_ahead).is_some());
622 bc.clear();
623 assert!(bc.needed_blocks(peer, count, best, 600, max_parallel, max_ahead).is_some());
624 assert_eq!(metrics.stale_download_reservations.get(), 1);
625 }
626
627 #[test]
628 fn gap_above_common_number_requested() {
629 let mut bc = BlockCollection::new();
630 assert!(is_empty(&bc));
631
632 let count = 5;
633 let best = 30;
634 let max_parallel = 1;
637 let max_ahead = 200;
638
639 let peer1 = PeerId::random();
640 let peer2 = PeerId::random();
641 let peer3 = PeerId::random();
642
643 let common = 10;
644 assert_eq!(
645 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
646 Some(11..16),
647 );
648 assert_eq!(
649 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
650 Some(16..21),
651 );
652 assert_eq!(
653 bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
654 Some(21..26),
655 );
656
657 bc.clear_peer_download(&peer2);
660
661 assert_eq!(
663 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
664 Some(16..21),
665 );
666 }
667
668 #[test]
669 fn gap_below_common_number_not_requested() {
670 let mut bc = BlockCollection::new();
671 assert!(is_empty(&bc));
672
673 let count = 5;
674 let best = 30;
675 let max_parallel = 1;
678 let max_ahead = 200;
679
680 let peer1 = PeerId::random();
681 let peer2 = PeerId::random();
682 let peer3 = PeerId::random();
683
684 let common = 10;
685 assert_eq!(
686 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
687 Some(11..16),
688 );
689 assert_eq!(
690 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
691 Some(16..21),
692 );
693 assert_eq!(
694 bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
695 Some(21..26),
696 );
697
698 bc.clear_peer_download(&peer2);
701
702 let common = 23;
704 assert_eq!(
705 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
706 Some(26..31), );
708 }
709
710 #[test]
711 fn range_at_the_end_above_common_number_requested() {
712 let mut bc = BlockCollection::new();
713 assert!(is_empty(&bc));
714
715 let count = 5;
716 let best = 30;
717 let max_parallel = 1;
718 let max_ahead = 200;
719
720 let peer1 = PeerId::random();
721 let peer2 = PeerId::random();
722
723 let common = 10;
724 assert_eq!(
725 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
726 Some(11..16),
727 );
728 assert_eq!(
729 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
730 Some(16..21),
731 );
732 }
733
734 #[test]
735 fn range_at_the_end_below_common_number_not_requested() {
736 let mut bc = BlockCollection::new();
737 assert!(is_empty(&bc));
738
739 let count = 5;
740 let best = 30;
741 let max_parallel = 1;
742 let max_ahead = 200;
743
744 let peer1 = PeerId::random();
745 let peer2 = PeerId::random();
746
747 let common = 10;
748 assert_eq!(
749 bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
750 Some(11..16),
751 );
752
753 let common = 20;
754 assert_eq!(
755 bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
756 Some(21..26), );
758 }
759}