dist: decentralize download status

Author: djc

src/cli/self_update.rs

@@ -1275,14 +1275,7 @@ pub(crate) async fn prepare_update(dl_cfg: &DownloadCfg<'_>) -> Result<Option<Pa
 
     // Download new version
     info!("downloading self-update (new version: {available_version})");
-    download_file(
-        &download_url,
-        &setup_path,
-        None,
-        &dl_cfg.tracker,
-        dl_cfg.process,
-    )
-    .await?;
+    download_file(&download_url, &setup_path, None, None, dl_cfg.process).await?;
 
     // Mark as executable
     utils::make_executable(&setup_path)?;
@@ -1301,14 +1294,7 @@ async fn get_available_rustup_version(dl_cfg: &DownloadCfg<'_>) -> Result<String
     let release_file_url = format!("{update_root}/release-stable.toml");
     let release_file_url = utils::parse_url(&release_file_url)?;
     let release_file = tempdir.path().join("release-stable.toml");
-    download_file(
-        &release_file_url,
-        &release_file,
-        None,
-        &dl_cfg.tracker,
-        dl_cfg.process,
-    )
-    .await?;
+    download_file(&release_file_url, &release_file, None, None, dl_cfg.process).await?;
     let release_toml_str = utils::read_file("rustup release", &release_file)?;
     let release_toml = toml::from_str::<RustupManifest>(&release_toml_str)
         .context("unable to parse rustup release file")?;

src/cli/self_update/windows.rs

@@ -277,7 +277,7 @@ pub(crate) async fn try_install_msvc(
         &visual_studio_url,
         &visual_studio,
         None,
-        &dl_cfg.tracker,
+        None,
         dl_cfg.process,
     )
     .await?;

src/dist/download.rs

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::borrow::Cow;
 use std::fs;
 use std::ops;
 use std::path::{Path, PathBuf};
@@ -42,7 +42,12 @@ impl<'a> DownloadCfg<'a> {
     /// Partial downloads are stored in `self.download_dir`, keyed by hash. If the
     /// target file already exists, then the hash is checked and it is returned
     /// immediately without re-downloading.
-    pub(crate) async fn download(&self, url: &Url, hash: &str) -> Result<File> {
+    pub(crate) async fn download(
+        &self,
+        url: &Url,
+        hash: &str,
+        status: &DownloadStatus,
+    ) -> Result<File> {
         utils::ensure_dir_exists("Download Directory", self.download_dir)?;
         let target_file = self.download_dir.join(Path::new(hash));
 
@@ -76,7 +81,7 @@ impl<'a> DownloadCfg<'a> {
             &partial_file_path,
             Some(&mut hasher),
             true,
-            &self.tracker,
+            Some(status),
             self.process,
         )
         .await
@@ -125,7 +130,7 @@ impl<'a> DownloadCfg<'a> {
         let hash_url = utils::parse_url(&(url.to_owned() + ".sha256"))?;
         let hash_file = self.tmp_cx.new_file()?;
 
-        download_file(&hash_url, &hash_file, None, &self.tracker, self.process).await?;
+        download_file(&hash_url, &hash_file, None, None, self.process).await?;
 
         utils::read_file("hash", &hash_file).map(|s| s[0..64].to_owned())
     }
@@ -139,6 +144,7 @@ impl<'a> DownloadCfg<'a> {
         &self,
         url_str: &str,
         update_hash: Option<&Path>,
+        status: Option<&DownloadStatus>,
         ext: &str,
     ) -> Result<Option<(temp::File, String)>> {
         let hash = self.download_hash(url_str).await?;
@@ -166,7 +172,7 @@ impl<'a> DownloadCfg<'a> {
         let file = self.tmp_cx.new_file_with_ext("", ext)?;
 
         let mut hasher = Sha256::new();
-        download_file(&url, &file, Some(&mut hasher), &self.tracker, self.process).await?;
+        download_file(&url, &file, Some(&mut hasher), status, self.process).await?;
         let actual_hash = format!("{:x}", hasher.finalize());
 
         if hash != actual_hash {
@@ -192,8 +198,6 @@ impl<'a> DownloadCfg<'a> {
 pub(crate) struct DownloadTracker {
     /// MultiProgress bar for the downloads.
     multi_progress_bars: MultiProgress,
-    /// Mapping of URLs being downloaded to their corresponding download status.
-    file_progress_bars: Mutex<HashMap<String, DownloadStatus>>,
 }
 
 impl DownloadTracker {
@@ -207,70 +211,30 @@ impl DownloadTracker {
 
         Self {
             multi_progress_bars,
-            file_progress_bars: Mutex::new(HashMap::new()),
         }
     }
 
     /// Creates a new ProgressBar for the given component.
-    pub(crate) fn create_progress_bar(&self, component: String, url: String) {
+    pub(crate) fn status_for(&self, component: impl Into<Cow<'static, str>>) -> DownloadStatus {
         let status = DownloadStatus::new(component);
         self.multi_progress_bars.add(status.progress.clone());
-        self.file_progress_bars.lock().unwrap().insert(url, status);
-    }
-
-    /// Sets the length for a new ProgressBar and gives it a style.
-    pub(crate) fn content_length_received(&self, content_len: u64, url: &str) {
-        if let Some(status) = self.file_progress_bars.lock().unwrap().get(url) {
-            status.received_length(content_len);
-        }
-    }
-
-    /// Notifies self that data of size `len` has been received.
-    pub(crate) fn data_received(&self, len: usize, url: &str) {
-        let mut map = self.file_progress_bars.lock().unwrap();
-        if let Some(status) = map.get_mut(url) {
-            status.received_data(len);
-        };
-    }
-
-    /// Notifies self that the download has finished.
-    pub(crate) fn download_finished(&self, url: &str) {
-        let map = self.file_progress_bars.lock().unwrap();
-        if let Some(status) = map.get(url) {
-            status.finished()
-        };
-    }
-
-    /// Notifies self that the download has failed.
-    pub(crate) fn download_failed(&self, url: &str) {
-        let map = self.file_progress_bars.lock().unwrap();
-        if let Some(status) = map.get(url) {
-            status.failed();
-        };
-    }
-
-    /// Notifies self that the download is being retried.
-    pub(crate) fn retrying_download(&self, url: &str) {
-        let mut map = self.file_progress_bars.lock().unwrap();
-        if let Some(status) = map.get_mut(url) {
-            status.retrying();
-        };
+        status
     }
 }
 
-struct DownloadStatus {
+pub(crate) struct DownloadStatus {
     progress: ProgressBar,
     /// The instant where the download is being retried.
     ///
     /// Allows us to delay the reappearance of the progress bar so that the user can see
     /// the message "retrying download" for at least a second. Without it, the progress
     /// bar would reappear immediately, not allowing the user to correctly see the message,
     /// before the progress bar starts again.
-    retry_time: Option<Instant>,
+    retry_time: Mutex<Option<Instant>>,
 }
 
 impl DownloadStatus {
-    fn new(component: String) -> Self {
+    fn new(component: impl Into<Cow<'static, str>>) -> Self {
         let progress = ProgressBar::hidden();
         progress.set_style(
             ProgressStyle::with_template(
@@ -283,25 +247,23 @@ impl DownloadStatus {
 
         Self {
             progress,
-            retry_time: None,
+            retry_time: Mutex::new(None),
         }
     }
 
-    fn received_length(&self, len: u64) {
+    pub(crate) fn received_length(&self, len: u64) {
         self.progress.reset();
         self.progress.set_length(len);
     }
 
-    fn received_data(&mut self, len: usize) {
+    pub(crate) fn received_data(&self, len: usize) {
         self.progress.inc(len as u64);
-        if !self
-            .retry_time
-            .is_some_and(|instant| instant.elapsed() > Duration::from_secs(1))
-        {
+        let mut retry_time = self.retry_time.lock().unwrap();
+        if !retry_time.is_some_and(|instant| instant.elapsed() > Duration::from_secs(1)) {
             return;
         }
 
-        self.retry_time = None;
+        *retry_time = None;
         self.progress.set_style(
             ProgressStyle::with_template(
                 "{msg:>12.bold}  [{bar:40}] {bytes}/{total_bytes} ({bytes_per_sec}, ETA: {eta})",
@@ -311,24 +273,24 @@ impl DownloadStatus {
         );
     }
 
-    fn finished(&self) {
+    pub(crate) fn finished(&self) {
         self.progress.set_style(
             ProgressStyle::with_template("{msg:>12.bold}  downloaded {total_bytes} in {elapsed}")
                 .unwrap(),
         );
         self.progress.finish();
     }
 
-    fn failed(&self) {
+    pub(crate) fn failed(&self) {
         self.progress.set_style(
             ProgressStyle::with_template("{msg:>12.bold}  download failed after {elapsed}")
                 .unwrap(),
         );
         self.progress.finish();
     }
 
-    fn retrying(&mut self) {
-        self.retry_time = Some(Instant::now());
+    pub(crate) fn retrying(&self) {
+        *self.retry_time.lock().unwrap() = Some(Instant::now());
         self.progress
             .set_style(ProgressStyle::with_template("{msg:>12.bold}  retrying download").unwrap());
     }

src/dist/manifestation.rs

@@ -17,7 +17,7 @@ use crate::dist::component::{
     Components, Package, TarGzPackage, TarXzPackage, TarZStdPackage, Transaction,
 };
 use crate::dist::config::Config;
-use crate::dist::download::{DownloadCfg, File};
+use crate::dist::download::{DownloadCfg, DownloadStatus, File};
 use crate::dist::manifest::{Component, CompressionKind, HashedBinary, Manifest, TargetedPackage};
 use crate::dist::prefix::InstallPrefix;
 #[cfg(test)]
@@ -160,7 +160,15 @@ impl Manifestation {
         let mut things_downloaded = Vec::new();
         let components = update
             .components_urls_and_hashes(new_manifest)
-            .map(|res| res.map(|(component, binary)| ComponentBinary { component, binary }))
+            .map(|res| {
+                res.map(|(component, binary)| ComponentBinary {
+                    component,
+                    binary,
+                    status: download_cfg
+                        .tracker
+                        .status_for(component.short_name(new_manifest)),
+                })
+            })
             .collect::<Result<Vec<_>>>()?;
 
         let components_len = components.len();
@@ -179,13 +187,6 @@ impl Manifestation {
             .unwrap_or(DEFAULT_MAX_RETRIES);
 
         info!("downloading component(s)");
-        for bin in &components {
-            download_cfg.tracker.create_progress_bar(
-                bin.component.short_name(new_manifest),
-                bin.binary.url.clone(),
-            );
-        }
-
         let semaphore = Arc::new(Semaphore::new(concurrent_downloads));
         let component_stream = tokio_stream::iter(components.into_iter()).map(|bin| {
             let sem = semaphore.clone();
@@ -443,12 +444,9 @@ impl Manifestation {
             .unwrap()
             .replace(DEFAULT_DIST_SERVER, dl_cfg.tmp_cx.dist_server.as_str());
 
-        dl_cfg
-            .tracker
-            .create_progress_bar("rust".to_owned(), url.clone());
-
+        let status = dl_cfg.tracker.status_for("rust");
         let dl = dl_cfg
-            .download_and_check(&url, update_hash, ".tar.gz")
+            .download_and_check(&url, update_hash, Some(&status), ".tar.gz")
             .await?;
         if dl.is_none() {
             return Ok(None);
@@ -730,6 +728,7 @@ impl Update {
 struct ComponentBinary<'a> {
     component: &'a Component,
     binary: &'a HashedBinary,
+    status: DownloadStatus,
 }
 
 impl<'a> ComponentBinary<'a> {
@@ -744,13 +743,13 @@ impl<'a> ComponentBinary<'a> {
 
         let downloaded_file = RetryIf::spawn(
             FixedInterval::from_millis(0).take(max_retries),
-            || download_cfg.download(url, &self.binary.hash),
+            || download_cfg.download(url, &self.binary.hash, &self.status),
             |e: &anyhow::Error| {
                 // retry only known retriable cases
                 match e.downcast_ref::<RustupError>() {
                     Some(RustupError::BrokenPartialFile)
                     | Some(RustupError::DownloadingFile { .. }) => {
-                        download_cfg.tracker.retrying_download(url.as_str());
+                        self.status.retrying();
                         true
                     }
                     _ => false,

src/dist/manifestation/tests.rs

@@ -489,14 +489,7 @@ impl TestContext {
         // Download the dist manifest and place it into the installation prefix
         let manifest_url = make_manifest_url(&self.url, &self.toolchain)?;
         let manifest_file = self.tmp_cx.new_file()?;
-        download_file(
-            &manifest_url,
-            &manifest_file,
-            None,
-            &dl_cfg.tracker,
-            dl_cfg.process,
-        )
-        .await?;
+        download_file(&manifest_url, &manifest_file, None, None, dl_cfg.process).await?;
         let manifest_str = utils::read_file("manifest", &manifest_file)?;
         let manifest = Manifest::parse(&manifest_str)?;
 

src/dist/mod.rs

@@ -1236,7 +1236,7 @@ pub(crate) async fn dl_v2_manifest(
 ) -> Result<Option<(ManifestV2, String)>> {
     let manifest_url = toolchain.manifest_v2_url(dist_root, download.process);
     match download
-        .download_and_check(&manifest_url, update_hash, ".toml")
+        .download_and_check(&manifest_url, update_hash, None, ".toml")
         .await
     {
         Ok(manifest_dl) => {
@@ -1293,7 +1293,9 @@ async fn dl_v1_manifest(
     }
 
     let manifest_url = toolchain.manifest_v1_url(dist_root, download.process);
-    let manifest_dl = download.download_and_check(&manifest_url, None, "").await?;
+    let manifest_dl = download
+        .download_and_check(&manifest_url, None, None, "")
+        .await?;
     let (manifest_file, _) = manifest_dl.unwrap();
     let manifest_str = utils::read_file("manifest", &manifest_file)?;
     let urls = manifest_str