Move downloading crates higher up in Cargo

This commit is intended to lay the groundwork for downloading crates in parallel
from crates.io. It starts out by lifting up the download operation from deep
within the `Source` trait to the `PackageSet`. This should allow us to maintain
a queue of downloads specifically in a `PackageSet` and arbitrate access through
that one type, making it easier for us to implement parallel downloads.

The `Source` trait's `download` method may now fail saying "go download this
data", and then the data is fed back into the trait object once it's complete to
actually get the `Package` out.

src/cargo/core/package.rs

@@ -11,6 +11,7 @@ use lazycell::LazyCell;
 
 use core::{Dependency, Manifest, PackageId, SourceId, Target};
 use core::{FeatureMap, SourceMap, Summary};
+use core::source::MaybePackage;
 use core::interning::InternedString;
 use util::{internal, lev_distance, Config};
 use util::errors::{CargoResult, CargoResultExt};
@@ -240,16 +241,22 @@ impl hash::Hash for Package {
 pub struct PackageSet<'cfg> {
     packages: HashMap<PackageId, LazyCell<Package>>,
     sources: RefCell<SourceMap<'cfg>>,
+    config: &'cfg Config,
 }
 
 impl<'cfg> PackageSet<'cfg> {
-    pub fn new(package_ids: &[PackageId], sources: SourceMap<'cfg>) -> PackageSet<'cfg> {
+    pub fn new(
+        package_ids: &[PackageId],
+        sources: SourceMap<'cfg>,
+        config: &'cfg Config,
+    ) -> PackageSet<'cfg> {
         PackageSet {
             packages: package_ids
                 .iter()
                 .map(|id| (id.clone(), LazyCell::new()))
                 .collect(),
             sources: RefCell::new(sources),
+            config,
         }
     }
 
@@ -271,6 +278,14 @@ impl<'cfg> PackageSet<'cfg> {
         let pkg = source
             .download(id)
             .chain_err(|| format_err!("unable to get packages from source"))?;
+        let pkg = match pkg {
+            MaybePackage::Ready(pkg) => pkg,
+            MaybePackage::Download { url, descriptor } => {
+                self.config.shell().status("Downloading", descriptor)?;
+                let data = download(self.config, &url)?;
+                source.finish_download(id, data)?
+            }
+        };
         assert!(slot.fill(pkg).is_ok());
         Ok(slot.borrow().unwrap())
     }
@@ -279,3 +294,44 @@ impl<'cfg> PackageSet<'cfg> {
         self.sources.borrow()
     }
 }
+
+fn download(config: &Config, url: &str) -> CargoResult<Vec<u8>> {
+    use util::network;
+    use util::Progress;
+    use util::errors::HttpNot200;
+
+    let mut handle = config.http()?.borrow_mut();
+    handle.get(true)?;
+    handle.url(&url)?;
+    handle.follow_location(true)?;
+    let mut body = Vec::new();
+    network::with_retry(config, || {
+        body = Vec::new();
+        let mut pb = Progress::new("Fetch", config);
+        {
+            handle.progress(true)?;
+            let mut handle = handle.transfer();
+            handle.progress_function(|dl_total, dl_cur, _, _| {
+                pb.tick(dl_cur as usize, dl_total as usize).is_ok()
+            })?;
+            handle.write_function(|buf| {
+                body.extend_from_slice(buf);
+                Ok(buf.len())
+            })?;
+            handle.perform().chain_err(|| {
+                format!("failed to download from `{}`", url)
+            })?;
+        }
+        let code = handle.response_code()?;
+        if code != 200 && code != 0 {
+            let url = handle.effective_url()?.unwrap_or(&url);
+            Err(HttpNot200 {
+                code,
+                url: url.to_string(),
+            }.into())
+        } else {
+            Ok(())
+        }
+    })?;
+    Ok(body)
+}

src/cargo/core/registry.rs

@@ -37,6 +37,7 @@ pub trait Registry {
 /// a `Source`. Each `Source` in the map has been updated (using network
 /// operations if necessary) and is ready to be queried for packages.
 pub struct PackageRegistry<'cfg> {
+    config: &'cfg Config,
     sources: SourceMap<'cfg>,
 
     // A list of sources which are considered "overrides" which take precedent
@@ -81,6 +82,7 @@ impl<'cfg> PackageRegistry<'cfg> {
     pub fn new(config: &'cfg Config) -> CargoResult<PackageRegistry<'cfg>> {
         let source_config = SourceConfigMap::new(config)?;
         Ok(PackageRegistry {
+            config,
             sources: SourceMap::new(),
             source_ids: HashMap::new(),
             overrides: Vec::new(),
@@ -94,7 +96,7 @@ impl<'cfg> PackageRegistry<'cfg> {
 
     pub fn get(self, package_ids: &[PackageId]) -> PackageSet<'cfg> {
         trace!("getting packages; sources={}", self.sources.len());
-        PackageSet::new(package_ids, self.sources)
+        PackageSet::new(package_ids, self.sources, self.config)
     }
 
     fn ensure_loaded(&mut self, namespace: &SourceId, kind: Kind) -> CargoResult<()> {

src/cargo/core/source/mod.rs

@@ -49,7 +49,10 @@ pub trait Source {
 
     /// The download method fetches the full package for each name and
     /// version specified.
-    fn download(&mut self, package: &PackageId) -> CargoResult<Package>;
+    fn download(&mut self, package: &PackageId) -> CargoResult<MaybePackage>;
+
+    fn finish_download(&mut self, package: &PackageId, contents: Vec<u8>)
+        -> CargoResult<Package>;
 
     /// Generates a unique string which represents the fingerprint of the
     /// current state of the source.
@@ -74,6 +77,14 @@ pub trait Source {
     }
 }
 
+pub enum MaybePackage {
+    Ready(Package),
+    Download {
+        url: String,
+        descriptor: String,
+    }
+}
+
 impl<'a, T: Source + ?Sized + 'a> Source for Box<T> {
     /// Forwards to `Source::supports_checksums`
     fn supports_checksums(&self) -> bool {
@@ -111,10 +122,14 @@ impl<'a, T: Source + ?Sized + 'a> Source for Box<T> {
     }
 
     /// Forwards to `Source::download`
-    fn download(&mut self, id: &PackageId) -> CargoResult<Package> {
+    fn download(&mut self, id: &PackageId) -> CargoResult<MaybePackage> {
         (**self).download(id)
     }
 
+    fn finish_download(&mut self, id: &PackageId, data: Vec<u8>) -> CargoResult<Package> {
+        (**self).finish_download(id, data)
+    }
+
     /// Forwards to `Source::fingerprint`
     fn fingerprint(&self, pkg: &Package) -> CargoResult<String> {
         (**self).fingerprint(pkg)
@@ -126,6 +141,52 @@ impl<'a, T: Source + ?Sized + 'a> Source for Box<T> {
     }
 }
 
+impl<'a, T: Source + ?Sized + 'a> Source for &'a mut T {
+    fn supports_checksums(&self) -> bool {
+        (**self).supports_checksums()
+    }
+
+    fn requires_precise(&self) -> bool {
+        (**self).requires_precise()
+    }
+
+    fn query(&mut self, dep: &Dependency, f: &mut FnMut(Summary)) -> CargoResult<()> {
+        (**self).query(dep, f)
+    }
+
+    fn fuzzy_query(&mut self, dep: &Dependency, f: &mut FnMut(Summary)) -> CargoResult<()> {
+        (**self).fuzzy_query(dep, f)
+    }
+
+    fn source_id(&self) -> &SourceId {
+        (**self).source_id()
+    }
+
+    fn replaced_source_id(&self) -> &SourceId {
+        (**self).replaced_source_id()
+    }
+
+    fn update(&mut self) -> CargoResult<()> {
+        (**self).update()
+    }
+
+    fn download(&mut self, id: &PackageId) -> CargoResult<MaybePackage> {
+        (**self).download(id)
+    }
+
+    fn finish_download(&mut self, id: &PackageId, data: Vec<u8>) -> CargoResult<Package> {
+        (**self).finish_download(id, data)
+    }
+
+    fn fingerprint(&self, pkg: &Package) -> CargoResult<String> {
+        (**self).fingerprint(pkg)
+    }
+
+    fn verify(&self, pkg: &PackageId) -> CargoResult<()> {
+        (**self).verify(pkg)
+    }
+}
+
 /// A `HashMap` of `SourceId` -> `Box<Source>`
 #[derive(Default)]
 pub struct SourceMap<'src> {

src/cargo/ops/cargo_install.rs

@@ -12,6 +12,8 @@ use toml;
 
 use core::{Dependency, Edition, Package, PackageIdSpec, Source, SourceId};
 use core::{PackageId, Workspace};
+use core::source::SourceMap;
+use core::package::PackageSet;
 use core::compiler::{DefaultExecutor, Executor};
 use ops::{self, CompileFilter};
 use sources::{GitSource, PathSource, SourceConfigMap};
@@ -499,22 +501,28 @@ where
                 source.source_id(),
             )?;
             let deps = source.query_vec(&dep)?;
-            match deps.iter().map(|p| p.package_id()).max() {
+            let pkgid = match deps.iter().map(|p| p.package_id()).max() {
-                Some(pkgid) => {
+                Some(pkgid) => pkgid,
-                    let pkg = source.download(pkgid)?;
-                    Ok((pkg, Box::new(source)))
-                }
                 None => {
                     let vers_info = vers.map(|v| format!(" with version `{}`", v))
                         .unwrap_or_default();
-                    Err(format_err!(
+                    bail!(
                         "could not find `{}` in {}{}",
                         name,
                         source.source_id(),
                         vers_info
-                    ))
+                    )
                 }
-            }
+            };
+
+            let pkg = {
+                let mut map = SourceMap::new();
+                map.insert(Box::new(&mut source));
+                PackageSet::new(&[pkgid.clone()], map, config)
+                    .get(&pkgid)?
+                    .clone()
+            };
+            Ok((pkg, Box::new(source)))
         }
         None => {
             let candidates = list_all(&mut source)?;

src/cargo/sources/directory.rs

@@ -9,6 +9,7 @@ use hex;
 use serde_json;
 
 use core::{Dependency, Package, PackageId, Source, SourceId, Summary};
+use core::source::MaybePackage;
 use sources::PathSource;
 use util::{Config, Sha256};
 use util::errors::{CargoResult, CargoResultExt};
@@ -150,14 +151,19 @@ impl<'cfg> Source for DirectorySource<'cfg> {
         Ok(())
     }
 
-    fn download(&mut self, id: &PackageId) -> CargoResult<Package> {
+    fn download(&mut self, id: &PackageId) -> CargoResult<MaybePackage> {
         self.packages
             .get(id)
             .map(|p| &p.0)
             .cloned()
+            .map(MaybePackage::Ready)
             .ok_or_else(|| format_err!("failed to find package with id: {}", id))
     }
 
+    fn finish_download(&mut self, _id: &PackageId, _data: Vec<u8>) -> CargoResult<Package> {
+        panic!("no downloads to do")
+    }
+
     fn fingerprint(&self, pkg: &Package) -> CargoResult<String> {
         Ok(pkg.package_id().version().to_string())
     }

src/cargo/sources/git/source.rs

@@ -2,7 +2,7 @@ use std::fmt::{self, Debug, Formatter};
 
 use url::Url;
 
-use core::source::{Source, SourceId};
+use core::source::{Source, SourceId, MaybePackage};
 use core::GitReference;
 use core::{Dependency, Package, PackageId, Summary};
 use util::Config;
@@ -210,7 +210,7 @@ impl<'cfg> Source for GitSource<'cfg> {
         self.path_source.as_mut().unwrap().update()
     }
 
-    fn download(&mut self, id: &PackageId) -> CargoResult<Package> {
+    fn download(&mut self, id: &PackageId) -> CargoResult<MaybePackage> {
         trace!(
             "getting packages for package id `{}` from `{:?}`",
             id,
@@ -222,6 +222,10 @@ impl<'cfg> Source for GitSource<'cfg> {
             .download(id)
     }
 
+    fn finish_download(&mut self, _id: &PackageId, _data: Vec<u8>) -> CargoResult<Package> {
+        panic!("no download should have started")
+    }
+
     fn fingerprint(&self, _pkg: &Package) -> CargoResult<String> {
         Ok(self.rev.as_ref().unwrap().to_string())
     }

src/cargo/sources/path.rs

@@ -9,6 +9,7 @@ use ignore::Match;
 use ignore::gitignore::GitignoreBuilder;
 
 use core::{Dependency, Package, PackageId, Source, SourceId, Summary};
+use core::source::MaybePackage;
 use ops;
 use util::{self, internal, CargoResult};
 use util::paths;
@@ -540,14 +541,19 @@ impl<'cfg> Source for PathSource<'cfg> {
         Ok(())
     }
 
-    fn download(&mut self, id: &PackageId) -> CargoResult<Package> {
+    fn download(&mut self, id: &PackageId) -> CargoResult<MaybePackage> {
         trace!("getting packages; id={}", id);
 
         let pkg = self.packages.iter().find(|pkg| pkg.package_id() == id);
         pkg.cloned()
+            .map(MaybePackage::Ready)
             .ok_or_else(|| internal(format!("failed to find {} in path source", id)))
     }
 
+    fn finish_download(&mut self, _id: &PackageId, _data: Vec<u8>) -> CargoResult<Package> {
+        panic!("no download should have started")
+    }
+
     fn fingerprint(&self, pkg: &Package) -> CargoResult<String> {
         let (max, max_path) = self.last_modified_file(pkg)?;
         Ok(format!("{} ({})", max, max_path.display()))

src/cargo/sources/registry/local.rs

@@ -4,10 +4,9 @@ use std::path::Path;
 
 use core::PackageId;
 use hex;
-use sources::registry::{RegistryConfig, RegistryData};
+use sources::registry::{RegistryConfig, RegistryData, MaybeLock};
-use util::FileLock;
 use util::paths;
-use util::{Config, Filesystem, Sha256};
+use util::{Config, Filesystem, Sha256, FileLock};
 use util::errors::{CargoResult, CargoResultExt};
 
 pub struct LocalRegistry<'cfg> {
@@ -70,7 +69,7 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
         Ok(())
     }
 
-    fn download(&mut self, pkg: &PackageId, checksum: &str) -> CargoResult<FileLock> {
+    fn download(&mut self, pkg: &PackageId, checksum: &str) -> CargoResult<MaybeLock> {
         let crate_file = format!("{}-{}.crate", pkg.name(), pkg.version());
         let mut crate_file = self.root.open_ro(&crate_file, self.config, "crate file")?;
 
@@ -78,7 +77,7 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
         // checksum below as it is in theory already verified.
         let dst = format!("{}-{}", pkg.name(), pkg.version());
         if self.src_path.join(dst).into_path_unlocked().exists() {
-            return Ok(crate_file);
+            return Ok(MaybeLock::Ready(crate_file));
         }
 
         self.config.shell().status("Unpacking", pkg)?;
@@ -102,6 +101,12 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
 
         crate_file.seek(SeekFrom::Start(0))?;
 
-        Ok(crate_file)
+        Ok(MaybeLock::Ready(crate_file))
+    }
+
+    fn finish_download(&mut self, _pkg: &PackageId, _checksum: &str, _data: &[u8])
+        -> CargoResult<FileLock>
+    {
+        panic!("this source doesn't download")
     }
 }

src/cargo/sources/registry/mod.rs

@@ -170,6 +170,7 @@ use serde_json;
 use tar::Archive;
 
 use core::dependency::{Dependency, Kind};
+use core::source::MaybePackage;
 use core::{Package, PackageId, Source, SourceId, Summary};
 use sources::PathSource;
 use util::errors::CargoResultExt;
@@ -347,13 +348,20 @@ pub trait RegistryData {
     ) -> CargoResult<()>;
     fn config(&mut self) -> CargoResult<Option<RegistryConfig>>;
     fn update_index(&mut self) -> CargoResult<()>;
-    fn download(&mut self, pkg: &PackageId, checksum: &str) -> CargoResult<FileLock>;
+    fn download(&mut self, pkg: &PackageId, checksum: &str) -> CargoResult<MaybeLock>;
+    fn finish_download(&mut self, pkg: &PackageId, checksum: &str, data: &[u8])
+        -> CargoResult<FileLock>;
 
     fn is_crate_downloaded(&self, _pkg: &PackageId) -> bool {
         true
     }
 }
 
+pub enum MaybeLock {
+    Ready(FileLock),
+    Download { url: String, descriptor: String }
+}
+
 mod index;
 mod local;
 mod remote;
@@ -462,6 +470,34 @@ impl<'cfg> RegistrySource<'cfg> {
             index::RegistryIndex::new(&self.source_id, path, self.config, self.index_locked);
         Ok(())
     }
+
+    fn get_pkg(&mut self, package: &PackageId, path: FileLock) -> CargoResult<Package> {
+        let path = self
+            .unpack_package(package, &path)
+            .chain_err(|| internal(format!("failed to unpack package `{}`", package)))?;
+        let mut src = PathSource::new(&path, &self.source_id, self.config);
+        src.update()?;
+        let pkg = match src.download(package)? {
+            MaybePackage::Ready(pkg) => pkg,
+            MaybePackage::Download { .. } => unreachable!(),
+        };
+
+        // Unfortunately the index and the actual Cargo.toml in the index can
+        // differ due to historical Cargo bugs. To paper over these we trash the
+        // *summary* loaded from the Cargo.toml we just downloaded with the one
+        // we loaded from the index.
+        let summaries = self
+            .index
+            .summaries(package.name().as_str(), &mut *self.ops)?;
+        let summary = summaries
+            .iter()
+            .map(|s| &s.0)
+            .find(|s| s.package_id() == package)
+            .expect("summary not found");
+        let mut manifest = pkg.manifest().clone();
+        manifest.set_summary(summary.clone());
+        Ok(Package::new(manifest, pkg.manifest_path()))
+    }
 }
 
 impl<'cfg> Source for RegistrySource<'cfg> {
@@ -526,31 +562,24 @@ impl<'cfg> Source for RegistrySource<'cfg> {
         Ok(())
     }
 
-    fn download(&mut self, package: &PackageId) -> CargoResult<Package> {
+    fn download(&mut self, package: &PackageId) -> CargoResult<MaybePackage> {
         let hash = self.index.hash(package, &mut *self.ops)?;
-        let path = self.ops.download(package, &hash)?;
+        match self.ops.download(package, &hash)? {
-        let path = self
+            MaybeLock::Ready(file) => {
-            .unpack_package(package, &path)
+                self.get_pkg(package, file).map(MaybePackage::Ready)
-            .chain_err(|| internal(format!("failed to unpack package `{}`", package)))?;
+            }
-        let mut src = PathSource::new(&path, &self.source_id, self.config);
+            MaybeLock::Download { url, descriptor } => {
-        src.update()?;
+                Ok(MaybePackage::Download { url, descriptor })
-        let pkg = src.download(package)?;
+            }
+        }
+    }
 
-        // Unfortunately the index and the actual Cargo.toml in the index can
+    fn finish_download(&mut self, package: &PackageId, data: Vec<u8>)
-        // differ due to historical Cargo bugs. To paper over these we trash the
+        -> CargoResult<Package>
-        // *summary* loaded from the Cargo.toml we just downloaded with the one
+    {
-        // we loaded from the index.
+        let hash = self.index.hash(package, &mut *self.ops)?;
-        let summaries = self
+        let file = self.ops.finish_download(package, &hash, &data)?;
-            .index
+        self.get_pkg(package, file)
-            .summaries(package.name().as_str(), &mut *self.ops)?;
-        let summary = summaries
-            .iter()
-            .map(|s| &s.0)
-            .find(|s| s.package_id() == package)
-            .expect("summary not found");
-        let mut manifest = pkg.manifest().clone();
-        manifest.set_summary(summary.clone());
-        Ok(Package::new(manifest, pkg.manifest_path()))
     }
 
     fn fingerprint(&self, pkg: &Package) -> CargoResult<String> {

src/cargo/sources/registry/remote.rs

@@ -14,10 +14,10 @@ use lazycell::LazyCell;
 use core::{PackageId, SourceId};
 use sources::git;
 use sources::registry::{RegistryConfig, RegistryData, CRATE_TEMPLATE, INDEX_LOCK, VERSION_TEMPLATE};
-use util::network;
+use sources::registry::MaybeLock;
 use util::{FileLock, Filesystem};
-use util::{Config, Progress, Sha256, ToUrl};
+use util::{Config, Sha256};
-use util::errors::{CargoResult, CargoResultExt, HttpNot200};
+use util::errors::{CargoResult, CargoResultExt};
 
 pub struct RemoteRegistry<'cfg> {
     index_path: Filesystem,
@@ -122,6 +122,10 @@ impl<'cfg> RemoteRegistry<'cfg> {
         *self.tree.borrow_mut() = Some(tree);
         Ok(Ref::map(self.tree.borrow(), |s| s.as_ref().unwrap()))
     }
+
+    fn filename(&self, pkg: &PackageId) -> String {
+        format!("{}-{}.crate", pkg.name(), pkg.version())
+    }
 }
 
 impl<'cfg> RegistryData for RemoteRegistry<'cfg> {
@@ -206,9 +210,8 @@ impl<'cfg> RegistryData for RemoteRegistry<'cfg> {
         Ok(())
     }
 
-    fn download(&mut self, pkg: &PackageId, checksum: &str) -> CargoResult<FileLock> {
+    fn download(&mut self, pkg: &PackageId, _checksum: &str) -> CargoResult<MaybeLock> {
-        let filename = format!("{}-{}.crate", pkg.name(), pkg.version());
+        let filename = self.filename(pkg);
-        let path = Path::new(&filename);
 
         // Attempt to open an read-only copy first to avoid an exclusive write
         // lock and also work with read-only filesystems. Note that we check the
@@ -216,18 +219,12 @@ impl<'cfg> RegistryData for RemoteRegistry<'cfg> {
         //
         // If this fails then we fall through to the exclusive path where we may
         // have to redownload the file.
-        if let Ok(dst) = self.cache_path.open_ro(path, self.config, &filename) {
+        if let Ok(dst) = self.cache_path.open_ro(&filename, self.config, &filename) {
             let meta = dst.file().metadata()?;
             if meta.len() > 0 {
-                return Ok(dst);
+                return Ok(MaybeLock::Ready(dst));
             }
         }
-        let mut dst = self.cache_path.open_rw(path, self.config, &filename)?;
-        let meta = dst.file().metadata()?;
-        if meta.len() > 0 {
-            return Ok(dst);
-        }
-        self.config.shell().status("Downloading", pkg)?;
 
         let config = self.config()?.unwrap();
         let mut url = config.dl.clone();
@@ -235,56 +232,29 @@ impl<'cfg> RegistryData for RemoteRegistry<'cfg> {
             write!(url, "/{}/{}/download", CRATE_TEMPLATE, VERSION_TEMPLATE).unwrap();
         }
         let url = url.replace(CRATE_TEMPLATE, &*pkg.name())
-            .replace(VERSION_TEMPLATE, &pkg.version().to_string())
+            .replace(VERSION_TEMPLATE, &pkg.version().to_string());
-            .to_url()?;
 
-        // TODO: don't download into memory, but ensure that if we ctrl-c a
+        Ok(MaybeLock::Download { url, descriptor: pkg.to_string() })
-        //       download we should resume either from the start or the middle
+    }
-        //       on the next time
-        let url = url.to_string();
-        let mut handle = self.config.http()?.borrow_mut();
-        handle.get(true)?;
-        handle.url(&url)?;
-        handle.follow_location(true)?;
-        let mut state = Sha256::new();
-        let mut body = Vec::new();
-        network::with_retry(self.config, || {
-            state = Sha256::new();
-            body = Vec::new();
-            let mut pb = Progress::new("Fetch", self.config);
-            {
-                handle.progress(true)?;
-                let mut handle = handle.transfer();
-                handle.progress_function(|dl_total, dl_cur, _, _| {
-                    pb.tick(dl_cur as usize, dl_total as usize).is_ok()
-                })?;
-                handle.write_function(|buf| {
-                    state.update(buf);
-                    body.extend_from_slice(buf);
-                    Ok(buf.len())
-                })?;
-                handle.perform().chain_err(|| {
-                    format!("failed to download from `{}`", url)
-                })?;
-            }
-            let code = handle.response_code()?;
-            if code != 200 && code != 0 {
-                let url = handle.effective_url()?.unwrap_or(&url);
-                Err(HttpNot200 {
-                    code,
-                    url: url.to_string(),
-                }.into())
-            } else {
-                Ok(())
-            }
-        })?;
 
+    fn finish_download(&mut self, pkg: &PackageId, checksum: &str, data: &[u8])
+        -> CargoResult<FileLock>
+    {
         // Verify what we just downloaded
+        let mut state = Sha256::new();
+        state.update(data);
         if hex::encode(state.finish()) != checksum {
             bail!("failed to verify the checksum of `{}`", pkg)
         }
 
-        dst.write_all(&body)?;
+        let filename = self.filename(pkg);
+        let mut dst = self.cache_path.open_rw(&filename, self.config, &filename)?;
+        let meta = dst.file().metadata()?;
+        if meta.len() > 0 {
+            return Ok(dst);
+        }
+
+        dst.write_all(data)?;
         dst.seek(SeekFrom::Start(0))?;
         Ok(dst)
     }

src/cargo/sources/replaced.rs

@@ -1,4 +1,5 @@
 use core::{Dependency, Package, PackageId, Source, SourceId, Summary};
+use core::source::MaybePackage;
 use util::errors::{CargoResult, CargoResultExt};
 
 pub struct ReplacedSource<'cfg> {
@@ -71,11 +72,26 @@ impl<'cfg> Source for ReplacedSource<'cfg> {
         Ok(())
     }
 
-    fn download(&mut self, id: &PackageId) -> CargoResult<Package> {
+    fn download(&mut self, id: &PackageId) -> CargoResult<MaybePackage> {
         let id = id.with_source_id(&self.replace_with);
         let pkg = self.inner
             .download(&id)
             .chain_err(|| format!("failed to download replaced source {}", self.to_replace))?;
+        Ok(match pkg {
+            MaybePackage::Ready(pkg) => {
+                MaybePackage::Ready(pkg.map_source(&self.replace_with, &self.to_replace))
+            }
+            other @ MaybePackage::Download { .. } => other,
+        })
+    }
+
+    fn finish_download(&mut self, id: &PackageId, data: Vec<u8>)
+        -> CargoResult<Package>
+    {
+        let id = id.with_source_id(&self.replace_with);
+        let pkg = self.inner
+            .finish_download(&id, data)
+            .chain_err(|| format!("failed to download replaced source {}", self.to_replace))?;
         Ok(pkg.map_source(&self.replace_with, &self.to_replace))
     }
 

tests/testsuite/registry.rs

@@ -266,10 +266,7 @@ fn bad_cksum() {
             "\
 [UPDATING] [..] index
 [DOWNLOADING] bad-cksum [..]
-[ERROR] unable to get packages from source
+[ERROR] failed to download replaced source registry `https://[..]`
-
-Caused by:
-  failed to download replaced source registry `https://[..]`
 
 Caused by:
   failed to verify the checksum of `bad-cksum v0.0.1 (registry `[ROOT][..]`)`
@@ -1662,10 +1659,7 @@ fn bad_and_or_malicious_packages_rejected() {
             "\
 [UPDATING] [..]
 [DOWNLOADING] [..]
-error: unable to get packages from source
+error: failed to download [..]
-
-Caused by:
-  failed to download [..]
 
 Caused by:
   failed to unpack [..]