reth_downloaders/
file_client.rs

1use alloy_consensus::BlockHeader;
2use alloy_eips::BlockHashOrNumber;
3use alloy_primitives::{BlockHash, BlockNumber, Sealable, B256};
4use async_compression::tokio::bufread::GzipDecoder;
5use futures::Future;
6use itertools::{Either, Itertools};
7use reth_consensus::{Consensus, ConsensusError};
8use reth_network_p2p::{
9    bodies::client::{BodiesClient, BodiesFut},
10    download::DownloadClient,
11    error::RequestError,
12    headers::client::{HeadersClient, HeadersDirection, HeadersFut, HeadersRequest},
13    priority::Priority,
14    BlockClient,
15};
16use reth_network_peers::PeerId;
17use reth_primitives_traits::{Block, BlockBody, FullBlock, SealedBlock, SealedHeader};
18use std::{collections::HashMap, io, ops::RangeInclusive, path::Path, sync::Arc};
19use thiserror::Error;
20use tokio::{
21    fs::File,
22    io::{AsyncReadExt, BufReader},
23};
24use tokio_stream::StreamExt;
25use tokio_util::codec::FramedRead;
26use tracing::{debug, trace, warn};
27
28use super::file_codec::BlockFileCodec;
29use crate::receipt_file_client::FromReceiptReader;
30
31/// Default byte length of chunk to read from chain file.
32///
33/// Default is 1 GB.
34pub const DEFAULT_BYTE_LEN_CHUNK_CHAIN_FILE: u64 = 1_000_000_000;
35
36/// Front-end API for fetching chain data from a file.
37///
38/// Blocks are assumed to be written one after another in a file, as rlp bytes.
39///
40/// For example, if the file contains 3 blocks, the file is assumed to be encoded as follows:
41/// rlp(block1) || rlp(block2) || rlp(block3)
42///
43/// Blocks are assumed to have populated transactions, so reading headers will also buffer
44/// transactions in memory for use in the bodies stage.
45///
46/// This reads the entire file into memory, so it is not suitable for large files.
47#[derive(Debug, Clone)]
48pub struct FileClient<B: Block> {
49    /// The buffered headers retrieved when fetching new bodies.
50    headers: HashMap<BlockNumber, B::Header>,
51
52    /// A mapping between block hash and number.
53    hash_to_number: HashMap<BlockHash, BlockNumber>,
54
55    /// The buffered bodies retrieved when fetching new headers.
56    bodies: HashMap<BlockHash, B::Body>,
57}
58
59/// An error that can occur when constructing and using a [`FileClient`].
60#[derive(Debug, Error)]
61pub enum FileClientError {
62    /// An error occurred when validating a header from file.
63    #[error(transparent)]
64    Consensus(#[from] ConsensusError),
65
66    /// An error occurred when opening or reading the file.
67    #[error(transparent)]
68    Io(#[from] std::io::Error),
69
70    /// An error occurred when decoding blocks, headers, or rlp headers from the file.
71    #[error("{0}")]
72    Rlp(alloy_rlp::Error, Vec<u8>),
73
74    /// Custom error message.
75    #[error("{0}")]
76    Custom(&'static str),
77}
78
79impl From<&'static str> for FileClientError {
80    fn from(value: &'static str) -> Self {
81        Self::Custom(value)
82    }
83}
84
85impl<B: FullBlock> FileClient<B> {
86    /// Create a new file client from a slice of sealed blocks.
87    pub fn from_blocks(blocks: impl IntoIterator<Item = SealedBlock<B>>) -> Self {
88        let blocks: Vec<_> = blocks.into_iter().collect();
89        let capacity = blocks.len();
90
91        let mut headers = HashMap::with_capacity(capacity);
92        let mut hash_to_number = HashMap::with_capacity(capacity);
93        let mut bodies = HashMap::with_capacity(capacity);
94
95        for block in blocks {
96            let number = block.number();
97            let hash = block.hash();
98            let (header, body) = block.split_sealed_header_body();
99
100            headers.insert(number, header.into_header());
101            hash_to_number.insert(hash, number);
102            bodies.insert(hash, body);
103        }
104
105        Self { headers, hash_to_number, bodies }
106    }
107
108    /// Create a new file client from a file path.
109    pub async fn new<P: AsRef<Path>>(
110        path: P,
111        consensus: Arc<dyn Consensus<B>>,
112    ) -> Result<Self, FileClientError> {
113        let file = File::open(path).await?;
114        Self::from_file(file, consensus).await
115    }
116
117    /// Initialize the [`FileClient`] with a file directly.
118    pub(crate) async fn from_file(
119        mut file: File,
120        consensus: Arc<dyn Consensus<B>>,
121    ) -> Result<Self, FileClientError> {
122        // get file len from metadata before reading
123        let metadata = file.metadata().await?;
124        let file_len = metadata.len();
125
126        let mut reader = vec![];
127        file.read_to_end(&mut reader).await?;
128
129        Ok(FileClientBuilder { consensus, parent_header: None, skip_invalid_blocks: false }
130            .build(&reader[..], file_len)
131            .await?
132            .file_client)
133    }
134
135    /// Get the tip hash of the chain.
136    pub fn tip(&self) -> Option<B256> {
137        self.headers.get(&self.max_block()?).map(|h| h.hash_slow())
138    }
139
140    /// Get the start hash of the chain.
141    pub fn start(&self) -> Option<B256> {
142        self.headers.get(&self.min_block()?).map(|h| h.hash_slow())
143    }
144
145    /// Returns the highest block number of this client has or `None` if empty
146    pub fn max_block(&self) -> Option<u64> {
147        self.headers.keys().max().copied()
148    }
149
150    /// Returns the lowest block number of this client has or `None` if empty
151    pub fn min_block(&self) -> Option<u64> {
152        self.headers.keys().min().copied()
153    }
154
155    /// Clones and returns the highest header of this client has or `None` if empty. Seals header
156    /// before returning.
157    pub fn tip_header(&self) -> Option<SealedHeader<B::Header>> {
158        self.headers.get(&self.max_block()?).map(|h| SealedHeader::seal_slow(h.clone()))
159    }
160
161    /// Returns true if all blocks are canonical (no gaps)
162    pub fn has_canonical_blocks(&self) -> bool {
163        if self.headers.is_empty() {
164            return true
165        }
166        let (min, max) = self.headers.keys().minmax().into_option().expect("not empty");
167        // Contiguous range from min to max means no gaps
168        *max - *min + 1 == self.headers.len() as u64
169    }
170
171    /// Use the provided bodies as the file client's block body buffer.
172    pub fn with_bodies(mut self, bodies: HashMap<BlockHash, B::Body>) -> Self {
173        self.bodies = bodies;
174        self
175    }
176
177    /// Use the provided headers as the file client's block body buffer.
178    pub fn with_headers(mut self, headers: HashMap<BlockNumber, B::Header>) -> Self {
179        self.headers = headers;
180        for (number, header) in &self.headers {
181            self.hash_to_number.insert(header.hash_slow(), *number);
182        }
183        self
184    }
185
186    /// Returns the current number of headers in the client.
187    pub fn headers_len(&self) -> usize {
188        self.headers.len()
189    }
190
191    /// Returns the current number of bodies in the client.
192    pub fn bodies_len(&self) -> usize {
193        self.bodies.len()
194    }
195
196    /// Returns an iterator over headers in the client.
197    pub fn headers_iter(&self) -> impl Iterator<Item = &B::Header> {
198        self.headers.values()
199    }
200
201    /// Returns a mutable iterator over bodies in the client.
202    ///
203    /// Panics, if file client headers and bodies are not mapping 1-1.
204    pub fn bodies_iter_mut(&mut self) -> impl Iterator<Item = (u64, &mut B::Body)> {
205        let bodies = &mut self.bodies;
206        let numbers = &self.hash_to_number;
207        bodies.iter_mut().map(|(hash, body)| (numbers[hash], body))
208    }
209
210    /// Returns the current number of transactions in the client.
211    pub fn total_transactions(&self) -> usize {
212        self.bodies.iter().fold(0, |acc, (_, body)| acc + body.transactions().len())
213    }
214}
215
216struct FileClientBuilder<B: Block> {
217    pub consensus: Arc<dyn Consensus<B>>,
218    pub parent_header: Option<SealedHeader<B::Header>>,
219    pub skip_invalid_blocks: bool,
220}
221
222impl<B: FullBlock<Header: reth_primitives_traits::BlockHeader>> FromReader
223    for FileClientBuilder<B>
224{
225    type Error = FileClientError;
226    type Output = FileClient<B>;
227
228    /// Initialize the [`FileClient`] from bytes that have been read from file.
229    fn build<R>(
230        &self,
231        reader: R,
232        num_bytes: u64,
233    ) -> impl Future<Output = Result<DecodedFileChunk<Self::Output>, Self::Error>>
234    where
235        R: AsyncReadExt + Unpin,
236    {
237        let mut headers = HashMap::default();
238        let mut hash_to_number = HashMap::default();
239        let mut bodies = HashMap::default();
240
241        // use with_capacity to make sure the internal buffer contains the entire chunk
242        let mut stream =
243            FramedRead::with_capacity(reader, BlockFileCodec::<B>::default(), num_bytes as usize);
244
245        trace!(target: "downloaders::file",
246            target_num_bytes=num_bytes,
247            capacity=stream.read_buffer().capacity(),
248            "init decode stream"
249        );
250
251        let mut remaining_bytes = vec![];
252
253        let mut log_interval = 0;
254        let mut log_interval_start_block = 0;
255
256        let mut parent_header = self.parent_header.clone();
257
258        async move {
259            while let Some(block_res) = stream.next().await {
260                let block = match block_res {
261                    Ok(block) => block,
262                    Err(FileClientError::Rlp(err, bytes)) => {
263                        trace!(target: "downloaders::file",
264                            %err,
265                            bytes_len=bytes.len(),
266                            "partial block returned from decoding chunk"
267                        );
268                        remaining_bytes = bytes;
269                        break
270                    }
271                    Err(err) => return Err(err),
272                };
273
274                let block = SealedBlock::seal_slow(block);
275
276                // Run consensus pre-checks. An invalid block here (e.g. mid-file in a
277                // BlockchainTest sequence that intentionally interleaves invalid block proposals
278                // with the valid chain) is not a hard failure: skip the block and keep decoding
279                // so the pipeline can still apply the valid prefix.
280                let validation =
281                    self.consensus.validate_header(block.sealed_header()).and_then(|_| {
282                        if let Some(parent) = &parent_header {
283                            self.consensus
284                                .validate_header_against_parent(block.sealed_header(), parent)?;
285                        }
286                        self.consensus.validate_block_pre_execution(&block)
287                    });
288                if let Err(err) = validation {
289                    if !self.skip_invalid_blocks {
290                        return Err(err.into())
291                    }
292                    warn!(target: "downloaders::file",
293                        block_number = block.number(),
294                        block_hash = %block.hash(),
295                        %err,
296                        "skipping invalid block while decoding file"
297                    );
298                    continue
299                }
300                if parent_header.is_some() {
301                    parent_header = Some(block.sealed_header().clone());
302                }
303
304                // add to the internal maps
305                let block_hash = block.hash();
306                let block_number = block.number();
307                let (header, body) = block.split_sealed_header_body();
308                headers.insert(block_number, header.unseal());
309                hash_to_number.insert(block_hash, block_number);
310                bodies.insert(block_hash, body);
311
312                if log_interval == 0 {
313                    trace!(target: "downloaders::file",
314                        block_number,
315                        "read first block"
316                    );
317                    log_interval_start_block = block_number;
318                } else if log_interval % 100_000 == 0 {
319                    trace!(target: "downloaders::file",
320                        blocks=?log_interval_start_block..=block_number,
321                        "read blocks from file"
322                    );
323                    log_interval_start_block = block_number + 1;
324                }
325                log_interval += 1;
326            }
327
328            trace!(target: "downloaders::file", blocks = headers.len(), "Initialized file client");
329
330            Ok(DecodedFileChunk {
331                file_client: FileClient { headers, hash_to_number, bodies },
332                remaining_bytes,
333                highest_block: None,
334            })
335        }
336    }
337}
338
339impl<B: FullBlock> HeadersClient for FileClient<B> {
340    type Header = B::Header;
341    type Output = HeadersFut<B::Header>;
342
343    fn get_headers_with_priority(
344        &self,
345        request: HeadersRequest,
346        _priority: Priority,
347    ) -> Self::Output {
348        // this just searches the buffer, and fails if it can't find the header
349        let mut headers = Vec::new();
350        trace!(target: "downloaders::file", request=?request, "Getting headers");
351
352        let start_num = match request.start {
353            BlockHashOrNumber::Hash(hash) => match self.hash_to_number.get(&hash) {
354                Some(num) => *num,
355                None => {
356                    warn!(%hash, "Could not find starting block number for requested header hash");
357                    return Box::pin(async move { Err(RequestError::BadResponse) })
358                }
359            },
360            BlockHashOrNumber::Number(num) => num,
361        };
362
363        let range = if request.limit == 1 {
364            Either::Left(start_num..start_num + 1)
365        } else {
366            match request.direction {
367                HeadersDirection::Rising => Either::Left(start_num..start_num + request.limit),
368                HeadersDirection::Falling => {
369                    Either::Right((start_num - request.limit + 1..=start_num).rev())
370                }
371            }
372        };
373
374        trace!(target: "downloaders::file", range=?range, "Getting headers with range");
375
376        for block_number in range {
377            match self.headers.get(&block_number).cloned() {
378                Some(header) => headers.push(header),
379                None => {
380                    warn!(number=%block_number, "Could not find header");
381                    return Box::pin(async move { Err(RequestError::BadResponse) })
382                }
383            }
384        }
385
386        Box::pin(async move { Ok((PeerId::default(), headers).into()) })
387    }
388}
389
390impl<B: FullBlock> BodiesClient for FileClient<B> {
391    type Body = B::Body;
392    type Output = BodiesFut<B::Body>;
393
394    fn get_block_bodies_with_priority_and_range_hint(
395        &self,
396        hashes: Vec<B256>,
397        _priority: Priority,
398        _range_hint: Option<RangeInclusive<u64>>,
399    ) -> Self::Output {
400        // this just searches the buffer, and fails if it can't find the block
401        let mut bodies = Vec::new();
402
403        // check if any are an error
404        // could unwrap here
405        for hash in hashes {
406            match self.bodies.get(&hash).cloned() {
407                Some(body) => bodies.push(body),
408                None => return Box::pin(async move { Err(RequestError::BadResponse) }),
409            }
410        }
411
412        Box::pin(async move { Ok((PeerId::default(), bodies).into()) })
413    }
414}
415
416impl<B: FullBlock> DownloadClient for FileClient<B> {
417    fn report_bad_message(&self, _peer_id: PeerId) {
418        trace!("Reported a bad message on a file client, the file may be corrupted or invalid");
419        // noop
420    }
421
422    fn num_connected_peers(&self) -> usize {
423        // no such thing as connected peers when we are just using a file
424        1
425    }
426}
427
428impl<B: FullBlock> BlockClient for FileClient<B> {
429    type Block = B;
430}
431
432/// File reader type for handling different compression formats.
433#[derive(Debug)]
434enum FileReader {
435    /// Regular uncompressed file with remaining byte tracking.
436    Plain { file: File, remaining_bytes: u64 },
437    /// Gzip compressed file.
438    Gzip { decoder: GzipDecoder<BufReader<File>>, eof: bool },
439}
440
441impl FileReader {
442    /// Read some data into the provided buffer, returning the number of bytes read.
443    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> {
444        match self {
445            Self::Plain { file, .. } => file.read(buf).await,
446            Self::Gzip { decoder, .. } => decoder.read(buf).await,
447        }
448    }
449
450    const fn is_eof(&self) -> bool {
451        match self {
452            Self::Plain { remaining_bytes, .. } => *remaining_bytes == 0,
453            Self::Gzip { eof, .. } => *eof,
454        }
455    }
456
457    /// Read next chunk from file. Returns the number of bytes read for plain files,
458    /// or a boolean indicating if data is available for gzip files.
459    async fn read_next_chunk(
460        &mut self,
461        chunk: &mut Vec<u8>,
462        chunk_byte_len: u64,
463    ) -> Result<Option<u64>, FileClientError> {
464        match self {
465            Self::Plain { .. } => self.read_plain_chunk(chunk, chunk_byte_len).await,
466            Self::Gzip { .. } => {
467                Ok((self.read_gzip_chunk(chunk, chunk_byte_len).await?)
468                    .then_some(chunk.len() as u64))
469            }
470        }
471    }
472
473    async fn read_plain_chunk(
474        &mut self,
475        chunk: &mut Vec<u8>,
476        chunk_byte_len: u64,
477    ) -> Result<Option<u64>, FileClientError> {
478        let Self::Plain { file, remaining_bytes } = self else {
479            unreachable!("read_plain_chunk should only be called on Plain variant")
480        };
481
482        if *remaining_bytes == 0 && chunk.is_empty() {
483            // eof
484            return Ok(None)
485        }
486
487        let chunk_target_len = chunk_byte_len.min(*remaining_bytes + chunk.len() as u64);
488        let old_bytes_len = chunk.len() as u64;
489
490        // calculate reserved space in chunk
491        let new_read_bytes_target_len = chunk_target_len - old_bytes_len;
492
493        // read new bytes from file
494        let prev_read_bytes_len = chunk.len();
495        chunk.extend(std::iter::repeat_n(0, new_read_bytes_target_len as usize));
496        let reader = &mut chunk[prev_read_bytes_len..];
497
498        // actual bytes that have been read
499        let new_read_bytes_len = file.read_exact(reader).await? as u64;
500        let next_chunk_byte_len = chunk.len();
501
502        // update remaining file length
503        *remaining_bytes -= new_read_bytes_len;
504
505        debug!(target: "downloaders::file",
506            max_chunk_byte_len=chunk_byte_len,
507            prev_read_bytes_len,
508            new_read_bytes_target_len,
509            new_read_bytes_len,
510            next_chunk_byte_len,
511            remaining_file_byte_len=*remaining_bytes,
512            "new bytes were read from file"
513        );
514
515        Ok(Some(next_chunk_byte_len as u64))
516    }
517
518    /// Read next chunk from gzipped file.
519    async fn read_gzip_chunk(
520        &mut self,
521        chunk: &mut Vec<u8>,
522        chunk_byte_len: u64,
523    ) -> Result<bool, FileClientError> {
524        let mut buffer = vec![0u8; 64 * 1024];
525        loop {
526            if chunk.len() >= chunk_byte_len as usize {
527                return Ok(true)
528            }
529
530            match self.read(&mut buffer).await {
531                Ok(0) => {
532                    let Self::Gzip { eof, .. } = self else { unreachable!() };
533                    *eof = true;
534                    return Ok(!chunk.is_empty())
535                }
536                Ok(n) => {
537                    chunk.extend_from_slice(&buffer[..n]);
538                }
539                Err(e) => return Err(e.into()),
540            }
541        }
542    }
543}
544
545/// Chunks file into several [`FileClient`]s.
546#[derive(Debug)]
547pub struct ChunkedFileReader {
548    /// File reader (either plain or gzip).
549    file: FileReader,
550    /// Bytes that have been read.
551    chunk: Vec<u8>,
552    /// Max bytes per chunk.
553    chunk_byte_len: u64,
554    /// Optionally, tracks highest decoded block number. Needed when decoding data that maps * to 1
555    /// with block number
556    highest_block: Option<u64>,
557}
558
559impl ChunkedFileReader {
560    /// Opens the file to import from given path. Returns a new instance. If no chunk byte length
561    /// is passed, chunks have [`DEFAULT_BYTE_LEN_CHUNK_CHAIN_FILE`] (one static file).
562    /// Automatically detects gzip files by extension (.gz, .gzip).
563    pub async fn new<P: AsRef<Path>>(
564        path: P,
565        chunk_byte_len: Option<u64>,
566    ) -> Result<Self, FileClientError> {
567        let path = path.as_ref();
568        let file = File::open(path).await?;
569        let chunk_byte_len = chunk_byte_len.unwrap_or(DEFAULT_BYTE_LEN_CHUNK_CHAIN_FILE);
570
571        Self::from_file(
572            file,
573            chunk_byte_len,
574            path.extension()
575                .and_then(|ext| ext.to_str())
576                .is_some_and(|ext| ["gz", "gzip"].contains(&ext)),
577        )
578        .await
579    }
580
581    /// Opens the file to import from given path. Returns a new instance.
582    pub async fn from_file(
583        file: File,
584        chunk_byte_len: u64,
585        is_gzip: bool,
586    ) -> Result<Self, FileClientError> {
587        let file_reader = if is_gzip {
588            FileReader::Gzip { decoder: GzipDecoder::new(BufReader::new(file)), eof: false }
589        } else {
590            let remaining_bytes = file.metadata().await?.len();
591            FileReader::Plain { file, remaining_bytes }
592        };
593
594        Ok(Self { file: file_reader, chunk: vec![], chunk_byte_len, highest_block: None })
595    }
596
597    /// Reads bytes from file and buffers as next chunk to decode. Returns byte length of next
598    /// chunk to read.
599    async fn read_next_chunk(&mut self) -> Result<Option<u64>, FileClientError> {
600        self.file.read_next_chunk(&mut self.chunk, self.chunk_byte_len).await
601    }
602
603    /// Read next chunk from file. Returns [`FileClient`] containing decoded chunk.
604    ///
605    /// For gzipped files, this method accumulates data until at least `chunk_byte_len` bytes
606    /// are available before processing. For plain files, it uses the original chunking logic.
607    pub async fn next_chunk<B: FullBlock>(
608        &mut self,
609        consensus: Arc<dyn Consensus<B>>,
610        parent_header: Option<SealedHeader<B::Header>>,
611    ) -> Result<Option<FileClient<B>>, FileClientError> {
612        self.next_chunk_with_invalid_block_handling(consensus, parent_header, false).await
613    }
614
615    /// Read next chunk from file, optionally skipping blocks that fail consensus pre-checks.
616    pub async fn next_chunk_with_invalid_block_handling<B: FullBlock>(
617        &mut self,
618        consensus: Arc<dyn Consensus<B>>,
619        parent_header: Option<SealedHeader<B::Header>>,
620        skip_invalid_blocks: bool,
621    ) -> Result<Option<FileClient<B>>, FileClientError> {
622        let Some(chunk_len) = self.read_next_chunk().await? else { return Ok(None) };
623
624        // make new file client from chunk
625        let DecodedFileChunk { file_client, remaining_bytes, .. } =
626            FileClientBuilder { consensus, parent_header, skip_invalid_blocks }
627                .build(&self.chunk[..], chunk_len)
628                .await?;
629
630        if self.file.is_eof() && !remaining_bytes.is_empty() {
631            return Err(FileClientError::Rlp(alloy_rlp::Error::InputTooShort, remaining_bytes))
632        }
633
634        // save left over bytes
635        self.chunk = remaining_bytes;
636
637        Ok(Some(file_client))
638    }
639
640    /// Read next chunk from file. Returns [`FileClient`] containing decoded chunk.
641    pub async fn next_receipts_chunk<T>(&mut self) -> Result<Option<T>, T::Error>
642    where
643        T: FromReceiptReader,
644    {
645        let Some(next_chunk_byte_len) = self.read_next_chunk().await.map_err(|e| {
646            T::Error::from(match e {
647                FileClientError::Io(io_err) => io_err,
648                _ => io::Error::other(e.to_string()),
649            })
650        })?
651        else {
652            return Ok(None)
653        };
654
655        // make new file client from chunk
656        let DecodedFileChunk { file_client, remaining_bytes, highest_block } =
657            T::from_receipt_reader(&self.chunk[..], next_chunk_byte_len, self.highest_block)
658                .await?;
659
660        // save left over bytes
661        self.chunk = remaining_bytes;
662        // update highest block
663        self.highest_block = highest_block;
664
665        Ok(Some(file_client))
666    }
667}
668
669/// Constructs a file client from a reader.
670pub trait FromReader {
671    /// Error returned by file client type.
672    type Error: From<io::Error>;
673
674    /// Output returned by file client type.
675    type Output;
676
677    /// Returns a file client
678    fn build<R>(
679        &self,
680        reader: R,
681        num_bytes: u64,
682    ) -> impl Future<Output = Result<DecodedFileChunk<Self::Output>, Self::Error>>
683    where
684        Self: Sized,
685        R: AsyncReadExt + Unpin;
686}
687
688/// Output from decoding a file chunk with [`FromReader::build`].
689#[derive(Debug)]
690pub struct DecodedFileChunk<T> {
691    /// File client, i.e. the decoded part of chunk.
692    pub file_client: T,
693    /// Remaining bytes that have not been decoded, e.g. a partial block or a partial receipt.
694    pub remaining_bytes: Vec<u8>,
695    /// Highest block of decoded chunk. This is needed when decoding data that maps * to 1 with
696    /// block number, like receipts.
697    pub highest_block: Option<u64>,
698}
699
700#[cfg(test)]
701mod tests {
702    use super::*;
703    use crate::{
704        bodies::{
705            bodies::BodiesDownloaderBuilder,
706            test_utils::{insert_headers, zip_blocks},
707        },
708        headers::{reverse_headers::ReverseHeadersDownloaderBuilder, test_utils::child_header},
709        test_utils::{generate_bodies, generate_bodies_file},
710    };
711    use assert_matches::assert_matches;
712    use async_compression::tokio::write::GzipEncoder;
713    use futures_util::stream::StreamExt;
714    use rand::Rng;
715    use reth_consensus::{noop::NoopConsensus, test_utils::TestConsensus, ConsensusError};
716    use reth_ethereum_primitives::Block;
717    use reth_network_p2p::{
718        bodies::downloader::BodyDownloader,
719        headers::downloader::{HeaderDownloader, SyncTarget},
720    };
721    use reth_provider::test_utils::create_test_provider_factory;
722    use std::sync::Arc;
723    use tokio::{
724        fs::File,
725        io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, SeekFrom},
726    };
727
728    #[tokio::test]
729    async fn streams_bodies_from_buffer() {
730        // Generate some random blocks
731        let factory = create_test_provider_factory();
732        let (headers, mut bodies) = generate_bodies(0..=19);
733
734        insert_headers(&factory, &headers);
735
736        // create an empty file
737        let file = tempfile::tempfile().unwrap();
738
739        let client: Arc<FileClient<Block>> = Arc::new(
740            FileClient::from_file(file.into(), NoopConsensus::arc())
741                .await
742                .unwrap()
743                .with_bodies(bodies.clone().into_iter().collect()),
744        );
745        let mut downloader = BodiesDownloaderBuilder::default().build::<Block, _, _>(
746            client.clone(),
747            Arc::new(TestConsensus::default()),
748            factory,
749        );
750        downloader.set_download_range(0..=19).expect("failed to set download range");
751
752        assert_matches!(
753            downloader.next().await,
754            Some(Ok(res)) => assert_eq!(res, zip_blocks(headers.iter(), &mut bodies))
755        );
756    }
757
758    #[tokio::test]
759    async fn download_headers_at_fork_head() {
760        reth_tracing::init_test_tracing();
761
762        let p3 = SealedHeader::default();
763        let p2 = child_header(&p3);
764        let p1 = child_header(&p2);
765        let p0 = child_header(&p1);
766
767        let file = tempfile::tempfile().unwrap();
768        let client: Arc<FileClient<Block>> = Arc::new(
769            FileClient::from_file(file.into(), NoopConsensus::arc()).await.unwrap().with_headers(
770                HashMap::from([
771                    (0u64, p0.clone_header()),
772                    (1, p1.clone_header()),
773                    (2, p2.clone_header()),
774                    (3, p3.clone_header()),
775                ]),
776            ),
777        );
778
779        let mut downloader = ReverseHeadersDownloaderBuilder::default()
780            .stream_batch_size(3)
781            .request_limit(3)
782            .build(Arc::clone(&client), Arc::new(TestConsensus::default()));
783        downloader.update_local_head(p3.clone());
784        downloader.update_sync_target(SyncTarget::Tip(p0.hash()));
785
786        let headers = downloader.next().await.unwrap();
787        assert_eq!(headers.unwrap(), vec![p0, p1, p2]);
788        assert!(downloader.next().await.is_none());
789        assert!(downloader.next().await.is_none());
790    }
791
792    #[tokio::test]
793    async fn test_download_headers_from_file() {
794        reth_tracing::init_test_tracing();
795
796        // Generate some random blocks
797        let (file, headers, _) = generate_bodies_file(0..=19).await;
798        // now try to read them back
799        let client: Arc<FileClient<Block>> =
800            Arc::new(FileClient::from_file(file, NoopConsensus::arc()).await.unwrap());
801
802        // construct headers downloader and use first header
803        let mut header_downloader = ReverseHeadersDownloaderBuilder::default()
804            .build(Arc::clone(&client), Arc::new(TestConsensus::default()));
805        header_downloader.update_local_head(headers.first().unwrap().clone());
806        header_downloader.update_sync_target(SyncTarget::Tip(headers.last().unwrap().hash()));
807
808        // get headers first
809        let mut downloaded_headers = header_downloader.next().await.unwrap().unwrap();
810
811        // reverse to make sure it's in the right order before comparing
812        downloaded_headers.reverse();
813
814        // the first header is not included in the response
815        assert_eq!(downloaded_headers, headers[1..]);
816    }
817
818    #[tokio::test]
819    async fn test_download_bodies_from_file() {
820        // Generate some random blocks
821        let factory = create_test_provider_factory();
822        let (file, headers, mut bodies) = generate_bodies_file(0..=19).await;
823
824        // now try to read them back
825        let client: Arc<FileClient<Block>> =
826            Arc::new(FileClient::from_file(file, NoopConsensus::arc()).await.unwrap());
827
828        // insert headers in db for the bodies downloader
829        insert_headers(&factory, &headers);
830
831        let mut downloader = BodiesDownloaderBuilder::default().build::<Block, _, _>(
832            client.clone(),
833            Arc::new(TestConsensus::default()),
834            factory,
835        );
836        downloader.set_download_range(0..=19).expect("failed to set download range");
837
838        assert_matches!(
839            downloader.next().await,
840            Some(Ok(res)) => assert_eq!(res, zip_blocks(headers.iter(), &mut bodies))
841        );
842    }
843
844    #[tokio::test]
845    async fn strict_chunk_decode_fails_on_invalid_block() {
846        let (file, _, _) = generate_bodies_file(0..=2).await;
847        let chunk_byte_len = file.metadata().await.unwrap().len();
848        let mut reader = ChunkedFileReader::from_file(file, chunk_byte_len, false).await.unwrap();
849        let consensus = Arc::new(TestConsensus::default());
850        consensus.set_fail_validation(true);
851
852        let err = reader.next_chunk::<Block>(consensus, None).await.unwrap_err();
853
854        assert_matches!(err, FileClientError::Consensus(ConsensusError::BaseFeeMissing));
855    }
856
857    #[tokio::test]
858    async fn lenient_chunk_decode_skips_invalid_blocks() {
859        let (file, _, _) = generate_bodies_file(0..=2).await;
860        let chunk_byte_len = file.metadata().await.unwrap().len();
861        let mut reader = ChunkedFileReader::from_file(file, chunk_byte_len, false).await.unwrap();
862        let consensus = Arc::new(TestConsensus::default());
863        consensus.set_fail_validation(true);
864
865        let client = reader
866            .next_chunk_with_invalid_block_handling::<Block>(consensus, None, true)
867            .await
868            .unwrap()
869            .unwrap();
870
871        assert_eq!(client.headers_len(), 0);
872        assert!(client.tip().is_none());
873    }
874
875    #[tokio::test]
876    async fn trailing_transaction_data_at_eof_is_rejected() {
877        let block = alloy_primitives::hex!(
878            "f902cef90259a01f1e77fa4e08a5ce98ba78db75ca1a4623c10e832eeff5a4a770e9d5048bfa95a01dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d4934794000000000000000000000000000000000000c0fea059eb85c4cc1486f67674192abb9fff7ae7f38f2ecdd3c87458ae2b8462eb5ca2a0a79a055a833e5c8e9364a9f6f06e1e01856d25a3cc21bad067cd94eb5cf9c7e9a0f78dfb743fbd92ade140711c8bbc542b5e307f0ab7984eff35d751969fe57efab901000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000080018401c9c3808252080c80a000000000000000000000000000000000000000000000000000000000000000008800000000000000000aa056e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b4218080a00000000000000000000000000000000000000000000000000000000000000000a0e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855f86eb86c02f8680180830f4240830f424082520894000000000000000000000000000000000000c0de8080c001a079be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798a03d1613cba75c9e7513aee78156909bdb32050830f2faa3125e8fd64b5778be3000c0c0"
879        );
880        let mut file = File::from_std(tempfile::tempfile().unwrap());
881        file.write_all(&block).await.unwrap();
882        file.seek(SeekFrom::Start(0)).await.unwrap();
883        let mut reader =
884            ChunkedFileReader::from_file(file, block.len() as u64, false).await.unwrap();
885
886        let err = reader.next_chunk::<Block>(NoopConsensus::arc(), None).await.unwrap_err();
887
888        assert_matches!(
889            err,
890            FileClientError::Rlp(alloy_rlp::Error::InputTooShort, bytes) if bytes == block
891        );
892    }
893
894    #[tokio::test]
895    async fn test_chunk_download_headers_from_file() {
896        reth_tracing::init_test_tracing();
897
898        // Generate some random blocks
899        let (file, headers, _) = generate_bodies_file(0..=14).await;
900
901        // calculate min for chunk byte length range, pick a lower bound that guarantees at least
902        // one block will be read
903        let chunk_byte_len = rand::rng().random_range(2000..=10_000);
904        trace!(target: "downloaders::file::test", chunk_byte_len);
905
906        // init reader
907        let mut reader =
908            ChunkedFileReader::from_file(file, chunk_byte_len as u64, false).await.unwrap();
909
910        let mut downloaded_headers: Vec<SealedHeader> = vec![];
911
912        let mut local_header = headers.first().unwrap().clone();
913
914        // test
915        while let Some(client) =
916            reader.next_chunk::<Block>(NoopConsensus::arc(), None).await.unwrap()
917        {
918            let sync_target = client.tip_header().unwrap();
919
920            let sync_target_hash = sync_target.hash();
921
922            // construct headers downloader and use first header
923            let mut header_downloader = ReverseHeadersDownloaderBuilder::default()
924                .build(Arc::new(client), Arc::new(TestConsensus::default()));
925            header_downloader.update_local_head(local_header.clone());
926            header_downloader.update_sync_target(SyncTarget::Tip(sync_target_hash));
927
928            // get headers first
929            let mut downloaded_headers_chunk = header_downloader.next().await.unwrap().unwrap();
930
931            // export new local header to outer scope
932            local_header = sync_target;
933
934            // reverse to make sure it's in the right order before comparing
935            downloaded_headers_chunk.reverse();
936            downloaded_headers.extend_from_slice(&downloaded_headers_chunk);
937        }
938
939        // the first header is not included in the response
940        assert_eq!(headers[1..], downloaded_headers);
941    }
942
943    #[tokio::test]
944    async fn test_chunk_download_headers_from_gzip_file() {
945        reth_tracing::init_test_tracing();
946
947        // Generate some random blocks
948        let (file, headers, _) = generate_bodies_file(0..=14).await;
949
950        // Create a gzipped version of the file
951        let gzip_temp_file = tempfile::NamedTempFile::new().unwrap();
952        let gzip_path = gzip_temp_file.path().to_owned();
953        drop(gzip_temp_file); // Close the file so we can write to it
954
955        // Read original file content first
956        let mut original_file = file;
957        original_file.seek(SeekFrom::Start(0)).await.unwrap();
958        let mut original_content = Vec::new();
959        original_file.read_to_end(&mut original_content).await.unwrap();
960
961        let mut gzip_file = File::create(&gzip_path).await.unwrap();
962        let mut encoder = GzipEncoder::new(&mut gzip_file);
963
964        // Write the original content through the gzip encoder
965        encoder.write_all(&original_content).await.unwrap();
966        encoder.shutdown().await.unwrap();
967        drop(gzip_file);
968
969        // Reopen the gzipped file for reading
970        let gzip_file = File::open(&gzip_path).await.unwrap();
971
972        // calculate min for chunk byte length range, pick a lower bound that guarantees at least
973        // one block will be read
974        let chunk_byte_len = rand::rng().random_range(2000..=10_000);
975        trace!(target: "downloaders::file::test", chunk_byte_len);
976
977        // init reader with gzip=true
978        let mut reader =
979            ChunkedFileReader::from_file(gzip_file, chunk_byte_len as u64, true).await.unwrap();
980
981        let mut downloaded_headers: Vec<SealedHeader> = vec![];
982
983        let mut local_header = headers.first().unwrap().clone();
984
985        // test
986        while let Some(client) =
987            reader.next_chunk::<Block>(NoopConsensus::arc(), None).await.unwrap()
988        {
989            if client.headers_len() == 0 {
990                continue;
991            }
992
993            let sync_target = client.tip_header().expect("tip_header should not be None");
994
995            let sync_target_hash = sync_target.hash();
996
997            // construct headers downloader and use first header
998            let mut header_downloader = ReverseHeadersDownloaderBuilder::default()
999                .build(Arc::new(client), Arc::new(TestConsensus::default()));
1000            header_downloader.update_local_head(local_header.clone());
1001            header_downloader.update_sync_target(SyncTarget::Tip(sync_target_hash));
1002
1003            // get headers first
1004            let mut downloaded_headers_chunk = header_downloader.next().await.unwrap().unwrap();
1005
1006            // export new local header to outer scope
1007            local_header = sync_target;
1008
1009            // reverse to make sure it's in the right order before comparing
1010            downloaded_headers_chunk.reverse();
1011            downloaded_headers.extend_from_slice(&downloaded_headers_chunk);
1012        }
1013
1014        // the first header is not included in the response
1015        assert_eq!(headers[1..], downloaded_headers);
1016    }
1017}
reth_downloaders/file_client.rs

reth_downloaders/
file_client.rs
