1use std::{
9 cmp::{max, min, Reverse},
10 collections::{BTreeMap, BinaryHeap, VecDeque},
11 time::Duration,
12};
13
14use futures::{FutureExt, TryFutureExt};
15use monero_oxide::{block::Block, transaction::Transaction};
16use tokio::{
17 task::{JoinHandle, JoinSet},
18 time::{interval, timeout, MissedTickBehavior},
19};
20use tower::{util::BoxCloneService, Service, ServiceExt};
21use tracing::{instrument, Instrument, Span};
22
23use cuprate_async_buffer::{BufferAppender, BufferStream};
24use cuprate_constants::block::MAX_BLOCK_HEIGHT_USIZE;
25use cuprate_p2p_core::{handles::ConnectionHandle, NetworkZone};
26use cuprate_pruning::PruningSeed;
27
28use crate::{
29 constants::{
30 BLOCK_DOWNLOADER_REQUEST_TIMEOUT, EMPTY_CHAIN_ENTRIES_BEFORE_TOP_ASSUMED, LONG_BAN,
31 MAX_BLOCK_BATCH_LEN, MAX_DOWNLOAD_FAILURES, MOST_RECENT_BATCH_WEIGHTS_FOR_BATCH_SIZE,
32 },
33 peer_set::{ClientDropGuard, PeerSetRequest, PeerSetResponse},
34};
35
36mod block_queue;
37mod chain_tracker;
38mod download_batch;
39mod request_chain;
40#[cfg(test)]
41mod tests;
42
43use block_queue::{BlockQueue, ReadyQueueBatch};
44pub use chain_tracker::ChainEntry;
45use chain_tracker::{BlocksToRetrieve, ChainTracker};
46use download_batch::download_batch_task;
47use request_chain::{initial_chain_search, request_chain_entry_from_peer};
48
49#[derive(Debug, Clone)]
51pub struct BlockBatch {
52 pub blocks: Vec<(Block, Vec<Transaction>)>,
54 pub size: usize,
56 pub peer_handle: ConnectionHandle,
58}
59
60#[derive(Debug, Copy, Clone, PartialOrd, PartialEq, Ord, Eq)]
62pub struct BlockDownloaderConfig {
63 pub buffer_bytes: usize,
66 pub in_progress_queue_bytes: usize,
68 pub check_client_pool_interval: Duration,
70 pub target_batch_bytes: usize,
72 pub initial_batch_len: usize,
74}
75
76pub struct BlockDownloaderHandle {
78 pub stream: BufferStream<BlockBatch>,
80 pub task: JoinHandle<()>,
82}
83
84#[derive(Debug, thiserror::Error)]
86pub(crate) enum BlockDownloadError {
87 #[error("A request to a peer timed out.")]
88 TimedOut,
89 #[error("The block buffer was closed.")]
90 BufferWasClosed,
91 #[error("The peers we requested data from did not have all the data.")]
92 PeerDidNotHaveRequestedData,
93 #[error("The peers response to a request was invalid.")]
94 PeersResponseWasInvalid,
95 #[error("The chain we are following is invalid.")]
96 ChainInvalid,
97 #[error("Failed to find a more advanced chain to follow")]
98 FailedToFindAChainToFollow,
99 #[error("The peer did not send any overlapping blocks, unknown start height.")]
100 PeerSentNoOverlappingBlocks,
101 #[error("Service error: {0}")]
102 ServiceError(#[from] tower::BoxError),
103}
104
105pub enum ChainSvcRequest<N: NetworkZone> {
107 CompactHistory,
109 FindFirstUnknown(Vec<[u8; 32]>),
111 CumulativeDifficulty,
113
114 ValidateEntries(VecDeque<ChainEntry<N>>, usize),
115}
116
117pub enum ChainSvcResponse<N: NetworkZone> {
119 CompactHistory {
121 block_ids: Vec<[u8; 32]>,
125 cumulative_difficulty: u128,
127 },
128 FindFirstUnknown(Option<(usize, usize)>),
132 CumulativeDifficulty(u128),
136
137 ValidateEntries {
138 valid: VecDeque<ChainEntry<N>>,
139 unknown: VecDeque<ChainEntry<N>>,
140 },
141}
142
143#[instrument(level = "error", skip_all, name = "block_downloader")]
153pub fn download_blocks<N: NetworkZone, C>(
154 peer_set: BoxCloneService<PeerSetRequest, PeerSetResponse<N>, tower::BoxError>,
155 our_chain_svc: C,
156 config: BlockDownloaderConfig,
157) -> BlockDownloaderHandle
158where
159 C: Service<ChainSvcRequest<N>, Response = ChainSvcResponse<N>, Error = tower::BoxError>
160 + Send
161 + 'static,
162 C::Future: Send + 'static,
163{
164 let (buffer_appender, buffer_stream) = cuprate_async_buffer::new_buffer(config.buffer_bytes);
165
166 let task = tokio::spawn(
167 BlockDownloader::new(peer_set, our_chain_svc, buffer_appender, config)
168 .run()
169 .inspect_err(|e| tracing::debug!("Error downloading blocks: {e}"))
170 .map(drop)
171 .instrument(Span::current()),
172 );
173
174 BlockDownloaderHandle {
175 stream: buffer_stream,
176 task,
177 }
178}
179
180struct BlockDownloader<N: NetworkZone, C> {
206 peer_set: BoxCloneService<PeerSetRequest, PeerSetResponse<N>, tower::BoxError>,
208
209 our_chain_svc: C,
211
212 most_recent_batch_sizes: BinaryHeap<Reverse<(usize, BatchSizeInformation)>>,
213
214 amount_of_empty_chain_entries: usize,
218
219 block_download_tasks: JoinSet<BlockDownloadTaskResponse<N>>,
221 #[expect(clippy::type_complexity)]
225 chain_entry_task: JoinSet<Result<(ClientDropGuard<N>, ChainEntry<N>), BlockDownloadError>>,
226
227 inflight_requests: BTreeMap<usize, BlocksToRetrieve<N>>,
231
232 failed_batches: BinaryHeap<Reverse<usize>>,
236
237 block_queue: BlockQueue,
238
239 config: BlockDownloaderConfig,
241}
242
243impl<N: NetworkZone, C> BlockDownloader<N, C>
244where
245 C: Service<ChainSvcRequest<N>, Response = ChainSvcResponse<N>, Error = tower::BoxError>
246 + Send
247 + 'static,
248 C::Future: Send + 'static,
249{
250 fn new(
252 peer_set: BoxCloneService<PeerSetRequest, PeerSetResponse<N>, tower::BoxError>,
253 our_chain_svc: C,
254 buffer_appender: BufferAppender<BlockBatch>,
255 config: BlockDownloaderConfig,
256 ) -> Self {
257 Self {
258 peer_set,
259 our_chain_svc,
260 most_recent_batch_sizes: BinaryHeap::new(),
261 amount_of_empty_chain_entries: 0,
262 block_download_tasks: JoinSet::new(),
263 chain_entry_task: JoinSet::new(),
264 inflight_requests: BTreeMap::new(),
265 block_queue: BlockQueue::new(buffer_appender),
266 failed_batches: BinaryHeap::new(),
267 config,
268 }
269 }
270
271 fn check_pending_peers(
273 &mut self,
274 chain_tracker: &mut ChainTracker<N>,
275 pending_peers: &mut BTreeMap<PruningSeed, Vec<ClientDropGuard<N>>>,
276 ) {
277 tracing::debug!("Checking if we can give any work to pending peers.");
278
279 for peers in pending_peers.values_mut() {
280 while let Some(peer) = peers.pop() {
281 if peer.info.handle.is_closed() {
282 continue;
284 }
285
286 let client = self.try_handle_free_client(chain_tracker, peer);
287 if let Some(peer) = client {
288 peers.push(peer);
291 break;
292 }
293 }
294 }
295 }
296
297 fn amount_of_blocks_to_request(&self) -> usize {
298 let biggest_batch = self
299 .most_recent_batch_sizes
300 .iter()
301 .max_by(|batch_1, batch_2| {
302 let av1 = batch_1.0 .1.byte_size / batch_1.0 .1.len;
303 let av2 = batch_2.0 .1.byte_size / batch_2.0 .1.len;
304
305 av1.cmp(&av2)
306 });
307
308 let Some(biggest_batch) = biggest_batch else {
309 return self.config.initial_batch_len;
310 };
311
312 calculate_next_block_batch_size(
313 biggest_batch.0 .1.byte_size,
314 biggest_batch.0 .1.len,
315 self.config.target_batch_bytes,
316 )
317 }
318
319 fn request_inflight_batch_again(
326 &mut self,
327 client: ClientDropGuard<N>,
328 ) -> Option<ClientDropGuard<N>> {
329 tracing::debug!(
330 "Requesting an inflight batch, current ready queue size: {}",
331 self.block_queue.size()
332 );
333
334 assert!(
335 !self.inflight_requests.is_empty(),
336 "We need requests inflight to be able to send the request again",
337 );
338
339 let oldest_ready_batch = self.block_queue.oldest_ready_batch().unwrap();
340
341 for (_, in_flight_batch) in self.inflight_requests.range_mut(0..oldest_ready_batch) {
342 if in_flight_batch.requests_sent >= 2 {
343 continue;
344 }
345
346 if !client_has_block_in_range(
347 &client.info.pruning_seed,
348 in_flight_batch.start_height,
349 in_flight_batch.ids.len(),
350 ) {
351 return Some(client);
352 }
353
354 self.block_download_tasks.spawn(download_batch_task(
355 client,
356 in_flight_batch.ids.clone(),
357 in_flight_batch.prev_id,
358 in_flight_batch.start_height,
359 in_flight_batch.requests_sent,
360 ));
361
362 return None;
363 }
364
365 tracing::debug!("Could not find an inflight request applicable for this peer.");
366
367 Some(client)
368 }
369
370 fn request_block_batch(
377 &mut self,
378 chain_tracker: &mut ChainTracker<N>,
379 client: ClientDropGuard<N>,
380 ) -> Option<ClientDropGuard<N>> {
381 tracing::trace!("Using peer to request a batch of blocks.");
382 while let Some(failed_request) = self.failed_batches.peek() {
384 let Some(request) = self.inflight_requests.get_mut(&failed_request.0) else {
387 self.failed_batches.pop();
389 continue;
390 };
391 if client_has_block_in_range(
393 &client.info.pruning_seed,
394 request.start_height,
395 request.ids.len(),
396 ) {
397 tracing::debug!("Using peer to request a failed batch");
398 request.requests_sent += 1;
401
402 self.block_download_tasks.spawn(download_batch_task(
403 client,
404 request.ids.clone(),
405 request.prev_id,
406 request.start_height,
407 request.requests_sent,
408 ));
409
410 self.failed_batches.pop();
412
413 return None;
414 }
415 break;
417 }
418
419 if self.block_queue.size() >= self.config.in_progress_queue_bytes {
421 return self.request_inflight_batch_again(client);
422 }
423
424 let Some(mut block_entry_to_get) = chain_tracker.blocks_to_get(
427 &client.info.pruning_seed,
428 self.amount_of_blocks_to_request(),
429 ) else {
430 return Some(client);
431 };
432
433 tracing::debug!("Requesting a new batch of blocks");
434
435 block_entry_to_get.requests_sent = 1;
436 self.inflight_requests
437 .insert(block_entry_to_get.start_height, block_entry_to_get.clone());
438
439 self.block_download_tasks.spawn(download_batch_task(
440 client,
441 block_entry_to_get.ids.clone(),
442 block_entry_to_get.prev_id,
443 block_entry_to_get.start_height,
444 block_entry_to_get.requests_sent,
445 ));
446
447 None
448 }
449
450 fn try_handle_free_client(
458 &mut self,
459 chain_tracker: &mut ChainTracker<N>,
460 client: ClientDropGuard<N>,
461 ) -> Option<ClientDropGuard<N>> {
462 if self.chain_entry_task.len() < 2
464 && self.amount_of_empty_chain_entries <= EMPTY_CHAIN_ENTRIES_BEFORE_TOP_ASSUMED
466 && chain_tracker.block_requests_queued(self.amount_of_blocks_to_request()) < 500
469 && chain_tracker.should_ask_for_next_chain_entry(&client.info.pruning_seed)
471 {
472 tracing::debug!("Requesting next chain entry");
473
474 let history = chain_tracker.get_simple_history();
475
476 self.chain_entry_task.spawn(
477 async move {
478 timeout(
479 BLOCK_DOWNLOADER_REQUEST_TIMEOUT,
480 request_chain_entry_from_peer(client, history),
481 )
482 .await
483 .map_err(|_| BlockDownloadError::TimedOut)?
484 }
485 .instrument(tracing::debug_span!(
486 "request_chain_entry",
487 current_height = chain_tracker.top_height()
488 )),
489 );
490
491 return None;
492 }
493
494 self.request_block_batch(chain_tracker, client)
496 }
497
498 async fn check_for_free_clients(
500 &mut self,
501 chain_tracker: &mut ChainTracker<N>,
502 pending_peers: &mut BTreeMap<PruningSeed, Vec<ClientDropGuard<N>>>,
503 ) -> Result<(), BlockDownloadError> {
504 tracing::debug!("Checking for free peers");
505
506 let ChainSvcResponse::CumulativeDifficulty(current_cumulative_difficulty) = self
508 .our_chain_svc
509 .ready()
510 .await?
511 .call(ChainSvcRequest::CumulativeDifficulty)
512 .await?
513 else {
514 panic!("Chain service returned wrong response.");
515 };
516
517 let PeerSetResponse::PeersWithMorePoW(clients) = self
518 .peer_set
519 .ready()
520 .await?
521 .call(PeerSetRequest::PeersWithMorePoW(
522 current_cumulative_difficulty,
523 ))
524 .await?
525 else {
526 unreachable!();
527 };
528
529 for client in clients {
530 pending_peers
531 .entry(client.info.pruning_seed)
532 .or_default()
533 .push(client);
534 }
535
536 self.check_pending_peers(chain_tracker, pending_peers);
537
538 Ok(())
539 }
540
541 async fn handle_download_batch_res(
543 &mut self,
544 start_height: usize,
545 res: Result<(ClientDropGuard<N>, BlockBatch), BlockDownloadError>,
546 chain_tracker: &mut ChainTracker<N>,
547 pending_peers: &mut BTreeMap<PruningSeed, Vec<ClientDropGuard<N>>>,
548 ) -> Result<(), BlockDownloadError> {
549 tracing::debug!("Handling block download response");
550
551 match res {
552 Err(e) => {
553 if matches!(e, BlockDownloadError::ChainInvalid) {
554 self.inflight_requests.get(&start_height).inspect(|entry| {
557 tracing::warn!(
558 "Received an invalid chain from peer: {}, exiting block downloader (it should be restarted).",
559 entry.peer_who_told_us
560 );
561 entry.peer_who_told_us_handle.ban_peer(LONG_BAN);
562 });
563
564 return Err(e);
565 }
566
567 if let Some(batch) = self.inflight_requests.get_mut(&start_height) {
569 tracing::debug!("Error downloading batch: {e}");
570
571 batch.failures += 1;
572 if batch.failures > MAX_DOWNLOAD_FAILURES {
573 tracing::debug!(
574 "Too many errors downloading blocks, stopping the block downloader."
575 );
576 return Err(BlockDownloadError::TimedOut);
577 }
578
579 self.failed_batches.push(Reverse(start_height));
580 }
581 }
582 Ok((client, block_batch)) => {
583 if self.inflight_requests.remove(&start_height).is_none() {
585 tracing::debug!("Already retrieved batch");
586 pending_peers
588 .entry(client.info.pruning_seed)
589 .or_default()
590 .push(client);
591
592 self.check_pending_peers(chain_tracker, pending_peers);
593
594 return Ok(());
595 }
596
597 if start_height
600 > self
601 .most_recent_batch_sizes
602 .peek()
603 .map(|Reverse((height, _))| *height)
604 .unwrap_or_default()
605 {
606 self.most_recent_batch_sizes.push(Reverse((
607 start_height,
608 BatchSizeInformation {
609 len: block_batch.blocks.len(),
610 byte_size: block_batch.size,
611 },
612 )));
613
614 if self.most_recent_batch_sizes.len() > MOST_RECENT_BATCH_WEIGHTS_FOR_BATCH_SIZE
615 {
616 self.most_recent_batch_sizes.pop();
617 }
618 }
619
620 self.block_queue
621 .add_incoming_batch(
622 ReadyQueueBatch {
623 start_height,
624 block_batch,
625 },
626 self.inflight_requests.first_key_value().map(|(k, _)| *k),
627 )
628 .await?;
629
630 pending_peers
631 .entry(client.info.pruning_seed)
632 .or_default()
633 .push(client);
634
635 self.check_pending_peers(chain_tracker, pending_peers);
636 }
637 }
638
639 Ok(())
640 }
641
642 async fn run(mut self) -> Result<(), BlockDownloadError> {
644 let mut chain_tracker =
645 initial_chain_search(&mut self.peer_set, &mut self.our_chain_svc).await?;
646
647 let mut pending_peers = BTreeMap::new();
648
649 tracing::info!("Attempting to download blocks from peers, this may take a while.");
650
651 let mut check_client_pool_interval = interval(self.config.check_client_pool_interval);
652 check_client_pool_interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
653
654 self.check_for_free_clients(&mut chain_tracker, &mut pending_peers)
655 .await?;
656
657 loop {
658 tokio::select! {
659 _ = check_client_pool_interval.tick() => {
660 tracing::debug!("Checking client pool for free peers, timer fired.");
661 self.check_for_free_clients(&mut chain_tracker, &mut pending_peers).await?;
662
663 if self.inflight_requests.is_empty() && self.amount_of_empty_chain_entries >= EMPTY_CHAIN_ENTRIES_BEFORE_TOP_ASSUMED {
665 tracing::debug!("Failed to find any more chain entries, probably fround the top");
666 return Ok(());
667 }
668 }
669 Some(res) = self.block_download_tasks.join_next() => {
670 let BlockDownloadTaskResponse {
671 start_height,
672 result
673 } = res.expect("Download batch future panicked");
674
675 self.handle_download_batch_res(start_height, result, &mut chain_tracker, &mut pending_peers).await?;
676
677 if self.inflight_requests.is_empty() && self.amount_of_empty_chain_entries >= EMPTY_CHAIN_ENTRIES_BEFORE_TOP_ASSUMED {
679 tracing::debug!("Failed to find any more chain entries, probably fround the top");
680 return Ok(());
681 }
682 }
683 Some(Ok(res)) = self.chain_entry_task.join_next() => {
684 match res {
685 Ok((client, entry)) => {
686 match chain_tracker.add_entry(entry, &mut self.our_chain_svc).await {
687 Ok(()) => {
688 tracing::debug!("Successfully added chain entry to chain tracker.");
689 self.amount_of_empty_chain_entries = 0;
690 }
691 Err(e) => {
692 tracing::debug!("Failed to add incoming chain entry to chain tracker: {e:?}");
693 self.amount_of_empty_chain_entries += 1;
694 }
695 }
696
697 pending_peers
698 .entry(client.info.pruning_seed)
699 .or_default()
700 .push(client);
701
702 self.check_pending_peers(&mut chain_tracker, &mut pending_peers);
703 }
704 Err(_) => self.amount_of_empty_chain_entries += 1
705 }
706 }
707 }
708 }
709 }
710}
711
712struct BlockDownloadTaskResponse<N: NetworkZone> {
714 start_height: usize,
716 result: Result<(ClientDropGuard<N>, BlockBatch), BlockDownloadError>,
718}
719
720const fn client_has_block_in_range(
722 pruning_seed: &PruningSeed,
723 start_height: usize,
724 length: usize,
725) -> bool {
726 pruning_seed.has_full_block(start_height, MAX_BLOCK_HEIGHT_USIZE)
727 && pruning_seed.has_full_block(start_height + length, MAX_BLOCK_HEIGHT_USIZE)
728}
729
730#[derive(Debug, Ord, PartialOrd, Eq, PartialEq)]
731struct BatchSizeInformation {
732 len: usize,
733 byte_size: usize,
734}
735
736fn calculate_next_block_batch_size(
743 previous_batch_size: usize,
744 previous_batch_len: usize,
745 target_batch_size: usize,
746) -> usize {
747 let adjusted_average_block_size = max((previous_batch_size * 2) / previous_batch_len, 1);
750
751 let next_batch_len = max(target_batch_size / adjusted_average_block_size, 1);
753
754 let next_batch_len = min(next_batch_len, (previous_batch_len * 3).div_ceil(2));
757
758 min(next_batch_len, MAX_BLOCK_BATCH_LEN)
760}