Skip to main content

cuprate_p2p/
block_downloader.rs

1//! # Block Downloader
2//!
3//! This module contains the block downloader, which finds a chain to
4//! download from our connected peers and downloads it. See the actual
5//! `struct` documentation for implementation details.
6//!
7//! The block downloader is started by [`download_blocks`].
8use 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/// A downloaded batch of blocks.
50#[derive(Debug, Clone)]
51pub struct BlockBatch {
52    /// The blocks.
53    pub blocks: Vec<(Block, Vec<Transaction>)>,
54    /// The size in bytes of this batch.
55    pub size: usize,
56    /// The peer that gave us this batch.
57    pub peer_handle: ConnectionHandle,
58}
59
60/// The block downloader config.
61#[derive(Debug, Copy, Clone, PartialOrd, PartialEq, Ord, Eq)]
62pub struct BlockDownloaderConfig {
63    /// The size in bytes of the buffer between the block downloader and the place which
64    /// is consuming the downloaded blocks.
65    pub buffer_bytes: usize,
66    /// The size of the in progress queue (in bytes) at which we stop requesting more blocks.
67    pub in_progress_queue_bytes: usize,
68    /// The [`Duration`] between checking the client pool for free peers.
69    pub check_client_pool_interval: Duration,
70    /// The target size of a single batch of blocks (in bytes).
71    pub target_batch_bytes: usize,
72    /// The initial amount of blocks to request (in number of blocks)
73    pub initial_batch_len: usize,
74}
75
76/// A running block downloader and its stream of downloaded blocks.
77pub struct BlockDownloaderHandle {
78    /// The sequential stream of downloaded blocks.
79    pub stream: BufferStream<BlockBatch>,
80    /// The downloader task.
81    pub task: JoinHandle<()>,
82}
83
84/// An error that occurred in the [`BlockDownloader`].
85#[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
105/// The request type for the chain service.
106pub enum ChainSvcRequest<N: NetworkZone> {
107    /// A request for the current chain history.
108    CompactHistory,
109    /// A request to find the first unknown block ID in a list of block IDs.
110    FindFirstUnknown(Vec<[u8; 32]>),
111    /// A request for our current cumulative difficulty.
112    CumulativeDifficulty,
113
114    ValidateEntries(VecDeque<ChainEntry<N>>, usize),
115}
116
117/// The response type for the chain service.
118pub enum ChainSvcResponse<N: NetworkZone> {
119    /// The response for [`ChainSvcRequest::CompactHistory`].
120    CompactHistory {
121        /// A list of blocks IDs in our chain, starting with the most recent block, all the way to the genesis block.
122        ///
123        /// These blocks should be in reverse chronological order, not every block is needed.
124        block_ids: Vec<[u8; 32]>,
125        /// The current cumulative difficulty of the chain.
126        cumulative_difficulty: u128,
127    },
128    /// The response for [`ChainSvcRequest::FindFirstUnknown`].
129    ///
130    /// Contains the index of the first unknown block and its expected height.
131    FindFirstUnknown(Option<(usize, usize)>),
132    /// The response for [`ChainSvcRequest::CumulativeDifficulty`].
133    ///
134    /// The current cumulative difficulty of our chain.
135    CumulativeDifficulty(u128),
136
137    ValidateEntries {
138        valid: VecDeque<ChainEntry<N>>,
139        unknown: VecDeque<ChainEntry<N>>,
140    },
141}
142
143/// This function starts the block downloader and returns a [`BlockDownloaderHandle`].
144///
145/// The block downloader will pick the longest chain and will follow it for as long as possible,
146/// the blocks given from the [`BufferStream`] will be in order.
147///
148/// The downloader runs until it is aborted with the returned [`JoinHandle`].
149///
150/// The block downloader may fail before the whole chain is downloaded. If this is the case you can
151/// call this function again, so it can start the search again.
152#[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
180/// # Block Downloader
181///
182/// This is the block downloader, which finds a chain to follow and attempts to follow it, adding the
183/// downloaded blocks to an [`async_buffer`].
184///
185/// ## Implementation Details
186///
187/// The first step to downloading blocks is to find a chain to follow, this is done by [`initial_chain_search`],
188/// docs can be found on that function for details on how this is done.
189///
190/// With an initial list of block IDs to follow the block downloader will then look for available peers
191/// to download blocks from.
192///
193/// For each peer we will then allocate a batch of blocks for them to retrieve, as these blocks come in
194/// we add them to the [`BlockQueue`] for pushing into the [`async_buffer`], once we have the oldest block downloaded
195/// we send it into the buffer, repeating this until the oldest current block is still being downloaded.
196///
197/// When a peer has finished downloading blocks we add it to our list of ready peers, so it can be used to
198/// request more data from.
199///
200/// Ready peers will either:
201/// - download the next batch of blocks
202/// - request the next chain entry
203/// - download an already requested batch of blocks (this might happen due to an error in the previous request
204///   or because the queue of ready blocks is too large, so we need the oldest block to clear it).
205struct BlockDownloader<N: NetworkZone, C> {
206    /// The peer set.
207    peer_set: BoxCloneService<PeerSetRequest, PeerSetResponse<N>, tower::BoxError>,
208
209    /// The service that holds our current chain state.
210    our_chain_svc: C,
211
212    most_recent_batch_sizes: BinaryHeap<Reverse<(usize, BatchSizeInformation)>>,
213
214    /// The amount of consecutive empty chain entries we received.
215    ///
216    /// An empty chain entry means we reached the peer's chain tip.
217    amount_of_empty_chain_entries: usize,
218
219    /// The running block download tasks.
220    block_download_tasks: JoinSet<BlockDownloadTaskResponse<N>>,
221    /// The running chain entry tasks.
222    ///
223    /// Returns a result of the chain entry or an error.
224    #[expect(clippy::type_complexity)]
225    chain_entry_task: JoinSet<Result<(ClientDropGuard<N>, ChainEntry<N>), BlockDownloadError>>,
226
227    /// The current inflight requests.
228    ///
229    /// This is a map of batch start heights to block IDs and related information of the batch.
230    inflight_requests: BTreeMap<usize, BlocksToRetrieve<N>>,
231
232    /// A queue of start heights from failed batches that should be retried.
233    ///
234    /// Wrapped in [`Reverse`] so we prioritize early batches.
235    failed_batches: BinaryHeap<Reverse<usize>>,
236
237    block_queue: BlockQueue,
238
239    /// The [`BlockDownloaderConfig`].
240    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    /// Creates a new [`BlockDownloader`]
251    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    /// Checks if we can make use of any peers that are currently pending requests.
272    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                    // Peer has disconnected, drop it.
283                    continue;
284                }
285
286                let client = self.try_handle_free_client(chain_tracker, peer);
287                if let Some(peer) = client {
288                    // This peer is ok however it does not have the data we currently need, this will only happen
289                    // because of its pruning seed so just skip over all peers with this pruning seed.
290                    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    /// Attempts to send another request for an inflight batch
320    ///
321    /// This function will find the batch(es) that we are waiting on to clear our ready queue and sends another request
322    /// for them.
323    ///
324    /// Returns the [`ClientDropGuard`] back if it doesn't have the batch according to its pruning seed.
325    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    /// Spawns a task to request blocks from the given peer.
371    ///
372    /// The batch requested will depend on our current state, failed batches will be prioritised.
373    ///
374    /// Returns the [`ClientDropGuard`] back if it doesn't have the data we currently need according
375    /// to its pruning seed.
376    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        // First look to see if we have any failed requests.
383        while let Some(failed_request) = self.failed_batches.peek() {
384            // Check if we still have the request that failed - another peer could have completed it after
385            // failure.
386            let Some(request) = self.inflight_requests.get_mut(&failed_request.0) else {
387                // We don't have the request in flight so remove the failure.
388                self.failed_batches.pop();
389                continue;
390            };
391            // Check if this peer has the blocks according to their pruning seed.
392            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                // They should have the blocks so send the re-request to this peer.
399
400                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                // Remove the failure, we have just handled it.
411                self.failed_batches.pop();
412
413                return None;
414            }
415            // The peer doesn't have the batch according to its pruning seed.
416            break;
417        }
418
419        // If our ready queue is too large send duplicate requests for the blocks we are waiting on.
420        if self.block_queue.size() >= self.config.in_progress_queue_bytes {
421            return self.request_inflight_batch_again(client);
422        }
423
424        // No failed requests that we can handle, request some new blocks.
425
426        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    /// Attempts to give work to a free client.
451    ///
452    /// This function will use our current state to decide if we should send a request for a chain entry
453    /// or if we should request a batch of blocks.
454    ///
455    /// Returns the [`ClientDropGuard`] back if it doesn't have the data we currently need according
456    /// to its pruning seed.
457    fn try_handle_free_client(
458        &mut self,
459        chain_tracker: &mut ChainTracker<N>,
460        client: ClientDropGuard<N>,
461    ) -> Option<ClientDropGuard<N>> {
462        // We send 2 requests, so if one of them is slow or doesn't have the next chain, we still have a backup.
463        if self.chain_entry_task.len() < 2
464            // If we have had too many failures then assume the tip has been found so no more chain entries.
465            && self.amount_of_empty_chain_entries <= EMPTY_CHAIN_ENTRIES_BEFORE_TOP_ASSUMED
466            // Check we have a big buffer of pending block IDs to retrieve, we don't want to be waiting around
467            // for a chain entry.
468            && chain_tracker.block_requests_queued(self.amount_of_blocks_to_request()) < 500
469            // Make sure this peer actually has the chain.
470            && 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        // Request a batch of blocks instead.
495        self.request_block_batch(chain_tracker, client)
496    }
497
498    /// Checks the [`ClientPool`] for free peers.
499    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        // This value might be slightly behind but that's ok.
507        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    /// Handles a response to a request to get blocks from a peer.
542    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                    // If the chain was invalid ban the peer who told us about it and error here to stop the
555                    // block downloader.
556                    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                // Add the request to the failed list.
568                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                // Remove the batch from the inflight batches.
584                if self.inflight_requests.remove(&start_height).is_none() {
585                    tracing::debug!("Already retrieved batch");
586                    // If it was already retrieved then there is nothing else to do.
587                    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 the batch is higher than the last time we updated `amount_of_blocks_to_request`, update it
598                // again.
599                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    /// Starts the main loop of the block downloader.
643    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 we have no inflight requests, and we have had too many empty chain entries in a row assume the top has been found.
664                    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 we have no inflight requests, and we have had too many empty chain entries in a row assume the top has been found.
678                    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
712/// The return value from the block download tasks.
713struct BlockDownloadTaskResponse<N: NetworkZone> {
714    /// The start height of the batch.
715    start_height: usize,
716    /// A result containing the batch or an error.
717    result: Result<(ClientDropGuard<N>, BlockBatch), BlockDownloadError>,
718}
719
720/// Returns if a peer has all the blocks in a range, according to its [`PruningSeed`].
721const 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
736/// Calculates the next amount of blocks to request in a batch.
737///
738/// Parameters:
739/// - `previous_batch_size` is the size, in bytes, of the last batch
740/// - `previous_batch_len` is the amount of blocks in the last batch
741/// - `target_batch_size` is the target size, in bytes, of a batch
742fn calculate_next_block_batch_size(
743    previous_batch_size: usize,
744    previous_batch_len: usize,
745    target_batch_size: usize,
746) -> usize {
747    // The average block size of the last batch of blocks, multiplied by 2 as a safety margin for
748    // future blocks.
749    let adjusted_average_block_size = max((previous_batch_size * 2) / previous_batch_len, 1);
750
751    // Set the amount of blocks to request equal to our target batch size divided by the adjusted_average_block_size.
752    let next_batch_len = max(target_batch_size / adjusted_average_block_size, 1);
753
754    // Cap the amount of growth to 1.5x the previous batch len, to prevent a small block causing us to request
755    // a huge amount of blocks.
756    let next_batch_len = min(next_batch_len, (previous_batch_len * 3).div_ceil(2));
757
758    // Cap the length to the maximum allowed.
759    min(next_batch_len, MAX_BLOCK_BATCH_LEN)
760}