reth_era_downloader/
client.rs

1use alloy_primitives::{hex, hex::ToHexExt};
2use bytes::Bytes;
3use eyre::{eyre, OptionExt};
4use futures_util::{stream::StreamExt, Stream, TryStreamExt};
5use reqwest::{Client, IntoUrl, Url};
6use reth_era::common::file_ops::EraFileType;
7use reth_fs_util::FsPathError;
8use sha2::{Digest, Sha256};
9use std::{collections::HashMap, future::Future, path::Path, str::FromStr, sync::Arc};
10use tokio::{
11    fs::{self, File},
12    io::{self, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWriteExt},
13    sync::Mutex,
14    try_join,
15};
16
17/// Downloaded index page filename
18const INDEX_HTML_FILE: &str = "index.html";
19
20/// Accesses the network over HTTP.
21pub trait HttpClient {
22    /// Makes an HTTP GET request to `url`. Returns a stream of response body bytes.
23    fn get<U: IntoUrl + Send + Sync>(
24        &self,
25        url: U,
26    ) -> impl Future<
27        Output = eyre::Result<impl Stream<Item = eyre::Result<Bytes>> + Send + Sync + Unpin>,
28    > + Send
29           + Sync;
30}
31
32impl HttpClient for Client {
33    async fn get<U: IntoUrl + Send + Sync>(
34        &self,
35        url: U,
36    ) -> eyre::Result<impl Stream<Item = eyre::Result<Bytes>> + Unpin> {
37        let response = Self::get(self, url).send().await?;
38
39        Ok(response.bytes_stream().map_err(|e| eyre::Error::new(e)))
40    }
41}
42
43/// An HTTP client with features for downloading ERA files from an external HTTP accessible
44/// endpoint.
45#[derive(Debug, Clone)]
46pub struct EraClient<Http> {
47    client: Http,
48    url: Url,
49    folder: Box<Path>,
50    era_type: EraFileType,
51    file_lists: Arc<Mutex<HashMap<EraFileType, FileList>>>,
52}
53
54impl<Http: HttpClient + Clone> EraClient<Http> {
55    const CHECKSUMS: &'static str = "checksums.txt";
56
57    /// Constructs [`EraClient`] using `client` to download from `url` into `folder`.
58    ///
59    /// The file type is auto-detected from the URL. Use
60    /// [`with_era_type`](Self::with_era_type) to override.
61    pub fn new(client: Http, url: Url, folder: impl Into<Box<Path>>) -> Self {
62        let era_type = EraFileType::from_url(url.as_str());
63        Self { client, url, folder: folder.into(), era_type, file_lists: Default::default() }
64    }
65
66    /// Override the auto-detected [`EraFileType`].
67    pub const fn with_era_type(mut self, era_type: EraFileType) -> Self {
68        self.era_type = era_type;
69        self
70    }
71
72    /// Performs a GET request on `url` and stores the response body into a file located within
73    /// the `folder`.
74    pub async fn download_to_file(&mut self, url: impl IntoUrl) -> eyre::Result<Box<Path>> {
75        let path = self.folder.to_path_buf();
76
77        let url = url.into_url()?;
78        let client = self.client.clone();
79        let file_name = url
80            .path_segments()
81            .ok_or_eyre("cannot-be-a-base")?
82            .next_back()
83            .ok_or_eyre("empty path segments")?;
84        let path = path.join(file_name);
85
86        if !self.is_downloaded(file_name, &path).await? {
87            let number = self
88                .file_name_to_number(file_name)
89                .ok_or_eyre("Cannot parse number from file name")?;
90
91            // Download to a temp path and rename in only on success, so an interrupted download
92            // never leaves a partial file that later looks complete.
93            let tmp_path = path.with_extension("tmp");
94
95            let mut tries = 1..3;
96            let mut actual_checksum: eyre::Result<_>;
97            loop {
98                actual_checksum = async {
99                    let mut file = File::create(&tmp_path).await?;
100                    let mut stream = client.get(url.clone()).await?;
101                    let mut hasher = Sha256::new();
102
103                    while let Some(item) = stream.next().await.transpose()? {
104                        io::copy(&mut item.as_ref(), &mut file).await?;
105                        hasher.update(item);
106                    }
107
108                    Ok(hasher.finalize().to_vec())
109                }
110                .await;
111
112                if actual_checksum.is_ok() || tries.next().is_none() {
113                    break;
114                }
115            }
116
117            if self.era_type.has_checksums() {
118                self.assert_checksum(number, actual_checksum?)
119                    .await
120                    .map_err(|e| eyre!("{e} for {file_name} at {}", path.display()))?;
121            } else {
122                // No checksum to validate against; surface a failed download before renaming.
123                actual_checksum?;
124            }
125
126            fs::rename(&tmp_path, &path).await?;
127        }
128
129        Ok(path.into_boxed_path())
130    }
131
132    /// Recovers index of file following the latest downloaded file from a different run.
133    pub async fn recover_index(&self) -> Option<usize> {
134        let mut max = None;
135
136        if let Ok(mut dir) = fs::read_dir(&self.folder).await {
137            while let Ok(Some(entry)) = dir.next_entry().await {
138                if let Some(name) = entry.file_name().to_str() &&
139                    self.is_matching_era_file(name) &&
140                    let Some(number) = self.file_name_to_number(name) &&
141                    (max.is_none() || matches!(max, Some(max) if number > max))
142                {
143                    max.replace(number + 1);
144                }
145            }
146        }
147
148        max
149    }
150
151    /// Deletes files that are outside-of the working range.
152    pub async fn delete_outside_range(&self, index: usize, max_files: usize) -> eyre::Result<()> {
153        let last = index + max_files;
154
155        if let Ok(mut dir) = fs::read_dir(&self.folder).await {
156            while let Ok(Some(entry)) = dir.next_entry().await {
157                if let Some(name) = entry.file_name().to_str() &&
158                    self.is_matching_era_file(name) &&
159                    let Some(number) = self.file_name_to_number(name) &&
160                    (number < index || number >= last)
161                {
162                    reth_fs_util::remove_file_if_exists(entry.path())?;
163                }
164            }
165        }
166
167        Ok(())
168    }
169
170    /// Returns a download URL for the file corresponding to `number`.
171    pub async fn url(&self, number: usize) -> eyre::Result<Option<Url>> {
172        Ok(self.number_to_file_name(number).await?.map(|name| self.url.join(&name)).transpose()?)
173    }
174
175    /// Returns the number of files in the `folder`.
176    pub async fn files_count(&self) -> usize {
177        let mut count = 0usize;
178
179        if let Ok(mut dir) = fs::read_dir(&self.folder).await {
180            while let Ok(Some(entry)) = dir.next_entry().await {
181                if let Some(ext) = entry.path().extension().and_then(|ext| ext.to_str()) &&
182                    self.era_type
183                        .extensions()
184                        .iter()
185                        .any(|valid| valid.trim_start_matches('.') == ext)
186                {
187                    count += 1;
188                }
189            }
190        }
191
192        count
193    }
194
195    /// Fetches the list of ERA1/ERA files from `url` and stores it in a file located within
196    /// `folder`.
197    /// For era files, checksum.txt file does not exist, so the checksum verification is
198    /// skipped.
199    ///
200    /// A successful fetch replaces the filename and checksum snapshot shared with this client's
201    /// clones. Later lookups use that snapshot until the next successful fetch; changes made to
202    /// sidecar files externally are not observed until then. Failed refreshes retain the last
203    /// successful snapshot. Archive types have separate snapshots. Refreshes are serialized across
204    /// clones, and lookups wait until an ongoing refresh finishes.
205    pub async fn fetch_file_list(&self) -> eyre::Result<()> {
206        let mut file_lists = self.file_lists.lock().await;
207        let index_path = self.folder.to_path_buf().join(INDEX_HTML_FILE);
208        let checksums_path = self.folder.to_path_buf().join(Self::CHECKSUMS);
209
210        // Only for files that ship checksums (era1, ere) we also download the checksums file.
211        if self.era_type.has_checksums() {
212            let checksums_url = self.url.join(Self::CHECKSUMS)?;
213            try_join!(
214                self.download_file_to_path(self.url.clone(), &index_path),
215                self.download_file_to_path(checksums_url, &checksums_path)
216            )?;
217        } else {
218            // Download only index file
219            self.download_file_to_path(self.url.clone(), &index_path).await?;
220        }
221
222        // Parse and extract era filenames from index.html
223        let file_names = self.extract_era_filenames(&index_path).await?;
224        let checksums = if self.era_type.has_checksums() {
225            fs::read(&checksums_path)
226                .await
227                .map_err(|error| FsPathError::read(error, &checksums_path))?
228                .split_inclusive(|byte| *byte == b'\n')
229                .map(|line| {
230                    let line = if let Some(line) = line.strip_suffix(b"\n") {
231                        line.strip_suffix(b"\r").unwrap_or(line)
232                    } else {
233                        line
234                    };
235                    line.to_vec()
236                })
237                .collect()
238        } else {
239            Vec::new()
240        };
241        file_lists.insert(self.era_type, FileList { file_names, checksums });
242        Ok(())
243    }
244
245    /// Extracts ERA filenames from `index.html` and writes them to the index file
246    async fn extract_era_filenames(&self, index_path: &Path) -> eyre::Result<Vec<String>> {
247        let file = File::open(index_path).await?;
248        let reader = io::BufReader::new(file);
249        let mut lines = reader.lines();
250
251        let path = self.folder.to_path_buf().join("index");
252        let file = File::create(&path).await?;
253        let mut writer = io::BufWriter::new(file);
254        let mut file_names = Vec::new();
255
256        while let Some(line) = lines.next_line().await? {
257            if let Some(era) = extract_era_filename(&line, self.era_type.extensions()) {
258                writer.write_all(era.as_bytes()).await?;
259                writer.write_all(b"\n").await?;
260                file_names.push(era.to_owned());
261            }
262        }
263
264        writer.flush().await?;
265        Ok(file_names)
266    }
267
268    // Helper to download a file to a specified path
269    async fn download_file_to_path(&self, url: Url, path: &Path) -> eyre::Result<()> {
270        let mut stream = self.client.get(url).await?;
271        let mut file = File::create(path).await?;
272
273        while let Some(item) = stream.next().await.transpose()? {
274            io::copy(&mut item.as_ref(), &mut file).await?;
275        }
276
277        Ok(())
278    }
279
280    /// Returns ERA1/ERA file name that is ordered at `number`.
281    pub async fn number_to_file_name(&self, number: usize) -> eyre::Result<Option<String>> {
282        if let Some(file_list) = self.file_lists.lock().await.get(&self.era_type) {
283            return Ok(file_list.file_names.get(number).cloned());
284        }
285        let path = self.folder.to_path_buf().join("index");
286        let file = File::open(&path).await?;
287        let reader = io::BufReader::new(file);
288        let mut lines = reader.lines();
289        for _ in 0..number {
290            lines.next_line().await?;
291        }
292
293        Ok(lines.next_line().await?)
294    }
295
296    async fn is_downloaded(&self, name: &str, path: impl AsRef<Path>) -> eyre::Result<bool> {
297        let path = path.as_ref();
298
299        match File::open(path).await {
300            Ok(file) => {
301                if self.era_type.has_checksums() {
302                    let number = self
303                        .file_name_to_number(name)
304                        .ok_or_else(|| eyre!("Cannot parse ERA number from {name}"))?;
305
306                    let actual_checksum = checksum(file).await?;
307                    let is_verified = self.verify_checksum(number, actual_checksum).await?;
308
309                    if !is_verified {
310                        fs::remove_file(path).await?;
311                    }
312
313                    Ok(is_verified)
314                } else {
315                    // For era files there is no checksums.txt, so verification is skipped.
316                    Ok(true)
317                }
318            }
319            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(false),
320            Err(e) => Err(e)?,
321        }
322    }
323
324    /// Returns `true` if `actual_checksum` matches expected checksum of the ERA1 file indexed by
325    /// `number` based on the [file list].
326    ///
327    /// [file list]: Self::fetch_file_list
328    async fn verify_checksum(&self, number: usize, actual_checksum: Vec<u8>) -> eyre::Result<bool> {
329        Ok(actual_checksum == self.expected_checksum(number).await?)
330    }
331
332    /// Returns `Ok` if `actual_checksum` matches expected checksum of the ERA1 file indexed by
333    /// `number` based on the [file list].
334    ///
335    /// [file list]: Self::fetch_file_list
336    async fn assert_checksum(&self, number: usize, actual_checksum: Vec<u8>) -> eyre::Result<()> {
337        let expected_checksum = self.expected_checksum(number).await?;
338
339        if actual_checksum == expected_checksum {
340            Ok(())
341        } else {
342            Err(eyre!(
343                "Checksum mismatch, got: {}, expected: {}",
344                actual_checksum.encode_hex(),
345                expected_checksum.encode_hex()
346            ))
347        }
348    }
349
350    /// Returns SHA-256 checksum for ERA1 file indexed by `number` based on the [file list].
351    ///
352    /// [file list]: Self::fetch_file_list
353    async fn expected_checksum(&self, number: usize) -> eyre::Result<Vec<u8>> {
354        if let Some(file_list) = self.file_lists.lock().await.get(&self.era_type) {
355            let checksum = file_list
356                .checksums
357                .get(number)
358                .ok_or_else(|| eyre!("Missing hash for number {number}"))?;
359            return Ok(hex::decode(checksum)?);
360        }
361        let file = File::open(self.folder.join(Self::CHECKSUMS)).await?;
362        let reader = io::BufReader::new(file);
363        let mut lines = reader.lines();
364
365        for _ in 0..number {
366            lines.next_line().await?;
367        }
368        let expected_checksum =
369            lines.next_line().await?.ok_or_else(|| eyre!("Missing hash for number {number}"))?;
370        let expected_checksum = hex::decode(expected_checksum)?;
371
372        Ok(expected_checksum)
373    }
374
375    fn file_name_to_number(&self, file_name: &str) -> Option<usize> {
376        file_name.split('-').nth(1).and_then(|v| usize::from_str(v).ok())
377    }
378
379    /// Whether `file_name` is a downloaded ERA file of this client's configured type.
380    ///
381    /// Excludes partial (`*.tmp`) and sidecar files that share the `<network>-<number>-...` stem,
382    /// so they don't influence resume or cleanup.
383    fn is_matching_era_file(&self, file_name: &str) -> bool {
384        EraFileType::from_filename(file_name) == Some(self.era_type)
385    }
386}
387
388/// The filenames and raw checksum lines from one successful file-list fetch.
389#[derive(Debug)]
390struct FileList {
391    file_names: Vec<String>,
392    checksums: Vec<Vec<u8>>,
393}
394
395/// Extracts an era filename ending in one of `extensions` from a single index line.
396///
397/// `extensions` are tried in order; pass them longest-first so `.ere` never matches inside `.erae`.
398fn extract_era_filename<'a>(line: &'a str, extensions: &[&str]) -> Option<&'a str> {
399    for ext in extensions {
400        // The separator can be multibyte, so the name starts after its full UTF-8 width.
401        if let Some(j) = line.find(ext) &&
402            let Some((i, sep)) =
403                line[..j].char_indices().rfind(|&(_, c)| !c.is_alphanumeric() && c != '-')
404        {
405            return Some(&line[i + sep.len_utf8()..j + ext.len()]);
406        }
407    }
408    None
409}
410
411async fn checksum(mut reader: impl AsyncRead + Unpin) -> eyre::Result<Vec<u8>> {
412    let mut hasher = Sha256::new();
413
414    // Create a buffer to read data into, sized for performance.
415    let mut data = vec![0; 64 * 1024];
416
417    loop {
418        // Read data from the reader into the buffer.
419        let len = reader.read(&mut data).await?;
420        if len == 0 {
421            break;
422        } // Exit loop if no more data.
423
424        // Update the hash with the data read.
425        hasher.update(&data[..len]);
426    }
427
428    // Finalize the hash after all data has been processed.
429    let hash = hasher.finalize().to_vec();
430
431    Ok(hash)
432}
433
434#[cfg(test)]
435mod tests {
436    use super::*;
437    use std::{
438        path::PathBuf,
439        sync::{
440            atomic::{AtomicBool, AtomicUsize, Ordering},
441            Arc,
442        },
443    };
444    use test_case::test_case;
445
446    impl EraClient<Client> {
447        fn empty() -> Self {
448            Self::new(Client::new(), Url::from_str("file:///").unwrap(), PathBuf::new())
449        }
450    }
451
452    #[test_case("mainnet-00600-a81ae85f.era1", Some(600))]
453    #[test_case("mainnet-00000-a81ae85f.era1", Some(0))]
454    #[test_case("00000-a81ae85f.era1", None)]
455    #[test_case("", None)]
456    fn test_file_name_to_number(file_name: &str, expected_number: Option<usize>) {
457        let client = EraClient::empty();
458
459        let actual_number = client.file_name_to_number(file_name);
460
461        assert_eq!(actual_number, expected_number);
462    }
463
464    // `.erae` lines must yield the full `.erae` name, never the `.ere` prefix inside it.
465    #[test_case(
466        "<a href=\"mainnet-00000-a6860fef.erae\">", &[".erae", ".ere"],
467        Some("mainnet-00000-a6860fef.erae"); "erae anchor not clipped to ere"
468    )]
469    #[test_case(
470        "    \"name\": \"mainnet-00001-05c64fc4.erae\",", &[".erae", ".ere"],
471        Some("mainnet-00001-05c64fc4.erae"); "erae json entry"
472    )]
473    #[test_case(
474        "<a href=\"mainnet-00600-a81ae85f.era1\">", &[".era1"],
475        Some("mainnet-00600-a81ae85f.era1"); "era1 anchor"
476    )]
477    #[test_case("<a href=\"checksums.txt\">", &[".erae", ".ere"], None; "no era file on line")]
478    #[test_case(
479        "<a href=\"\u{2026}mainnet-00000-a81ae85f.era1\">", &[".era1"],
480        Some("mainnet-00000-a81ae85f.era1"); "multibyte separator before name"
481    )]
482    fn test_extract_era_filename(line: &str, exts: &[&str], expected: Option<&str>) {
483        assert_eq!(extract_era_filename(line, exts), expected);
484    }
485
486    #[test]
487    fn test_with_era_type_overrides_auto_detection() {
488        // URL without "era1" auto-detects as Era
489        let client = EraClient::new(
490            Client::new(),
491            Url::from_str("https://example.com/").unwrap(),
492            PathBuf::new(),
493        );
494        assert_eq!(client.era_type, EraFileType::Era);
495
496        // with_era_type overrides to Era1
497        let client = client.with_era_type(EraFileType::Era1);
498        assert_eq!(client.era_type, EraFileType::Era1);
499    }
500
501    #[derive(Clone, Debug, Default)]
502    struct ListingClient {
503        index: Arc<Mutex<String>>,
504        checksums: Arc<Mutex<Vec<u8>>>,
505        fail: Arc<AtomicBool>,
506        calls: Arc<AtomicUsize>,
507    }
508
509    impl HttpClient for ListingClient {
510        async fn get<U: IntoUrl + Send + Sync>(
511            &self,
512            url: U,
513        ) -> eyre::Result<impl Stream<Item = eyre::Result<Bytes>> + Send + Sync + Unpin> {
514            self.calls.fetch_add(1, Ordering::Relaxed);
515            if self.fail.load(Ordering::Relaxed) {
516                return Err(eyre!("metadata fetch failed"));
517            }
518            let url = url.into_url()?;
519            let body = if url.path().ends_with("checksums.txt") {
520                self.checksums.lock().await.clone()
521            } else {
522                self.index.lock().await.clone().into_bytes()
523            };
524            Ok(futures_util::stream::iter([Ok(Bytes::from(body))]))
525        }
526    }
527
528    #[tokio::test]
529    async fn file_list_refresh_updates_clones_and_preserves_lazy_errors() {
530        let folder = tempfile::tempdir().unwrap();
531        let http = ListingClient::default();
532        *http.index.lock().await = "<a href=\"mainnet-00000-original.era1\">\n".to_owned();
533        *http.checksums.lock().await = b"00\r\n\xff\n".to_vec();
534        let client = EraClient::new(
535            http.clone(),
536            Url::parse("https://example.com/era1/").unwrap(),
537            folder.path(),
538        );
539        let clone = client.clone();
540        client.fetch_file_list().await.unwrap();
541        assert_eq!(
542            clone.number_to_file_name(0).await.unwrap().as_deref(),
543            Some("mainnet-00000-original.era1")
544        );
545        assert_eq!(clone.expected_checksum(0).await.unwrap(), vec![0]);
546        assert_eq!(
547            clone.expected_checksum(1).await.unwrap_err().downcast_ref::<hex::FromHexError>(),
548            Some(&hex::FromHexError::OddLength)
549        );
550
551        *http.index.lock().await =
552            "<a href=\"mainnet-00000-revised-name.era1\">\n<a href=\"mainnet-00001-next.era1\">\n"
553                .to_owned();
554        *http.checksums.lock().await = b"02\n03\n".to_vec();
555        http.fail.store(true, Ordering::Relaxed);
556        assert_eq!(
557            client.fetch_file_list().await.unwrap_err().to_string(),
558            "metadata fetch failed"
559        );
560        assert_eq!(
561            clone.number_to_file_name(0).await.unwrap().as_deref(),
562            Some("mainnet-00000-original.era1")
563        );
564        assert_eq!(clone.expected_checksum(0).await.unwrap(), vec![0]);
565
566        http.fail.store(false, Ordering::Relaxed);
567        client.fetch_file_list().await.unwrap();
568        assert_eq!(
569            clone.number_to_file_name(0).await.unwrap().as_deref(),
570            Some("mainnet-00000-revised-name.era1")
571        );
572        assert_eq!(
573            clone.number_to_file_name(1).await.unwrap().as_deref(),
574            Some("mainnet-00001-next.era1")
575        );
576        assert_eq!(clone.number_to_file_name(2).await.unwrap(), None);
577        assert_eq!(clone.expected_checksum(0).await.unwrap(), vec![2]);
578        assert_eq!(clone.expected_checksum(1).await.unwrap(), vec![3]);
579        assert_eq!(
580            clone.expected_checksum(2).await.unwrap_err().to_string(),
581            "Missing hash for number 2"
582        );
583    }
584
585    #[tokio::test]
586    async fn changing_archive_type_uses_a_separate_file_list() {
587        let folder = tempfile::tempdir().unwrap();
588        let http = ListingClient::default();
589        *http.index.lock().await = "<a href=\"mainnet-00000-original.era1\">\n".to_owned();
590        *http.checksums.lock().await = b"00\n".to_vec();
591        let client = EraClient::new(
592            http.clone(),
593            Url::parse("https://example.com/era1/").unwrap(),
594            folder.path(),
595        );
596        client.fetch_file_list().await.unwrap();
597        let era_client = client.clone().with_era_type(EraFileType::Era);
598        *http.index.lock().await = "<a href=\"mainnet-00000-consensus.era\">\n".to_owned();
599        era_client.fetch_file_list().await.unwrap();
600        assert_eq!(
601            era_client.number_to_file_name(0).await.unwrap().as_deref(),
602            Some("mainnet-00000-consensus.era")
603        );
604        assert_eq!(
605            client.number_to_file_name(0).await.unwrap().as_deref(),
606            Some("mainnet-00000-original.era1")
607        );
608        assert_eq!(client.expected_checksum(0).await.unwrap(), vec![0]);
609    }
610
611    #[tokio::test]
612    async fn lookups_without_fetch_use_existing_sidecars() {
613        let folder = tempfile::tempdir().unwrap();
614        reth_fs_util::write(folder.path().join("index"), b"first\nsecond-longer-name\n").unwrap();
615        reth_fs_util::write(folder.path().join(EraClient::<Client>::CHECKSUMS), b"00\n01\n")
616            .unwrap();
617        let client = EraClient::new(
618            Client::new(),
619            Url::parse("https://example.com/era1/").unwrap(),
620            folder.path(),
621        );
622        assert_eq!(
623            client.number_to_file_name(1).await.unwrap().as_deref(),
624            Some("second-longer-name")
625        );
626        assert_eq!(client.number_to_file_name(2).await.unwrap(), None);
627        assert_eq!(client.expected_checksum(1).await.unwrap(), vec![1]);
628        assert_eq!(
629            client.expected_checksum(2).await.unwrap_err().to_string(),
630            "Missing hash for number 2"
631        );
632    }
633
634    #[tokio::test]
635    async fn concurrent_refreshes_do_not_interleave_sidecars() {
636        let folder = tempfile::tempdir().unwrap();
637        let http = ListingClient::default();
638        *http.index.lock().await = "<a href=\"mainnet-00000-original.era1\">\n".to_owned();
639        *http.checksums.lock().await = b"00\n".to_vec();
640        let client = EraClient::new(
641            http.clone(),
642            Url::parse("https://example.com/era1/").unwrap(),
643            folder.path(),
644        );
645        let clone = client.clone();
646        let index_guard = http.index.lock().await;
647        let mut first = Box::pin(client.fetch_file_list());
648        assert!(futures::poll!(&mut first).is_pending());
649        assert_eq!(http.calls.load(Ordering::Relaxed), 2);
650        let mut second = Box::pin(clone.fetch_file_list());
651        assert!(futures::poll!(&mut second).is_pending());
652        assert_eq!(
653            http.calls.load(Ordering::Relaxed),
654            2,
655            "second refresh must wait before downloading sidecars"
656        );
657        drop(index_guard);
658        let (first, second) = tokio::join!(first, second);
659        first.unwrap();
660        second.unwrap();
661        assert_eq!(http.calls.load(Ordering::Relaxed), 4);
662        assert_eq!(clone.expected_checksum(0).await.unwrap(), vec![0]);
663    }
664}
reth_era_downloader/client.rs

reth_era_downloader/
client.rs
