Auto merge of #10482 - arlosi:refactor_load, r=Eh2406

Refactor RegistryData::load to handle management of the index cache

Enables registry implementations to signal if the cache is valid on a per-request basis.

Fixes a bug introduced by #10064 that caused Cargo not to update for several cases in a release build because it believed the index cache to be valid when it was not. The issue only occurred in release builds because debug builds verify that the cache contents is correct (by refreshing it).

Previously `current_version` was called by the index to determine whether the cache was valid. In the new model, `RegistryData::load` is passed the current version of the cache and returns an enum to indicate the status of the cached data.

r? `@eh2406`
cc `@ehuss`

Author: bors

src/cargo/sources/registry/index.rs

@@ -68,7 +68,7 @@
 
 use crate::core::dependency::Dependency;
 use crate::core::{PackageId, SourceId, Summary};
-use crate::sources::registry::{RegistryData, RegistryPackage, INDEX_V_MAX};
+use crate::sources::registry::{LoadResponse, RegistryData, RegistryPackage, INDEX_V_MAX};
 use crate::util::interning::InternedString;
 use crate::util::{internal, CargoResult, Config, Filesystem, OptVersionReq, ToSemver};
 use anyhow::bail;
@@ -247,6 +247,7 @@ pub struct IndexSummary {
 #[derive(Default)]
 struct SummariesCache<'a> {
     versions: Vec<(Version, &'a [u8])>,
+    index_version: &'a str,
 }
 
 impl<'cfg> RegistryIndex<'cfg> {
@@ -358,7 +359,6 @@ impl<'cfg> RegistryIndex<'cfg> {
 
         let root = load.assert_index_locked(&self.path);
         let cache_root = root.join(".cache");
-        let index_version = load.current_version();
 
         // See module comment in `registry/mod.rs` for why this is structured
         // the way it is.
@@ -376,7 +376,6 @@ impl<'cfg> RegistryIndex<'cfg> {
         // along the way produce helpful "did you mean?" suggestions.
         for (i, path) in UncanonicalizedIter::new(&raw_path).take(1024).enumerate() {
             let summaries = Summaries::parse(
-                index_version.as_deref(),
                 root,
                 &cache_root,
                 path.as_ref(),
@@ -559,7 +558,6 @@ impl Summaries {
     /// * `load` - the actual index implementation which may be very slow to
     ///   call. We avoid this if we can.
     pub fn parse(
-        index_version: Option<&str>,
         root: &Path,
         cache_root: &Path,
         relative: &Path,
@@ -571,88 +569,101 @@ impl Summaries {
         // of reasons, but consider all of them non-fatal and just log their
         // occurrence in case anyone is debugging anything.
         let cache_path = cache_root.join(relative);
-        let mut cache_contents = None;
-        if let Some(index_version) = index_version {
-            match fs::read(&cache_path) {
-                Ok(contents) => match Summaries::parse_cache(contents, index_version) {
-                    Ok(s) => {
-                        log::debug!("fast path for registry cache of {:?}", relative);
-                        if cfg!(debug_assertions) {
-                            cache_contents = Some(s.raw_data);
-                        } else {
-                            return Poll::Ready(Ok(Some(s)));
-                        }
-                    }
-                    Err(e) => {
-                        log::debug!("failed to parse {:?} cache: {}", relative, e);
-                    }
-                },
-                Err(e) => log::debug!("cache missing for {:?} error: {}", relative, e),
-            }
+        let mut cached_summaries = None;
+        let mut index_version = None;
+        match fs::read(&cache_path) {
+            Ok(contents) => match Summaries::parse_cache(contents) {
+                Ok((s, v)) => {
+                    cached_summaries = Some(s);
+                    index_version = Some(v);
+                }
+                Err(e) => {
+                    log::debug!("failed to parse {:?} cache: {}", relative, e);
+                }
+            },
+            Err(e) => log::debug!("cache missing for {:?} error: {}", relative, e),
         }
 
-        // This is the fallback path where we actually talk to the registry backend to load
-        // information. Here we parse every single line in the index (as we need
-        // to find the versions)
-        log::debug!("slow path for {:?}", relative);
+        let mut response = load.load(root, relative, index_version.as_deref())?;
+        // In debug builds, perform a second load without the cache so that
+        // we can validate that the cache is correct.
+        if cfg!(debug_assertions) && matches!(response, Poll::Ready(LoadResponse::CacheValid)) {
+            response = load.load(root, relative, None)?;
+        }
+        let response = match response {
+            Poll::Pending => return Poll::Pending,
+            Poll::Ready(response) => response,
+        };
+
+        let mut bytes_to_cache = None;
+        let mut version_to_cache = None;
         let mut ret = Summaries::default();
-        let mut hit_closure = false;
-        let mut cache_bytes = None;
-        let result = load.load(root, relative, &mut |contents| {
-            ret.raw_data = contents.to_vec();
-            let mut cache = SummariesCache::default();
-            hit_closure = true;
-            for line in split(contents, b'\n') {
-                // Attempt forwards-compatibility on the index by ignoring
-                // everything that we ourselves don't understand, that should
-                // allow future cargo implementations to break the
-                // interpretation of each line here and older cargo will simply
-                // ignore the new lines.
-                let summary = match IndexSummary::parse(config, line, source_id) {
-                    Ok(summary) => summary,
-                    Err(e) => {
-                        // This should only happen when there is an index
-                        // entry from a future version of cargo that this
-                        // version doesn't understand. Hopefully, those future
-                        // versions of cargo correctly set INDEX_V_MAX and
-                        // CURRENT_CACHE_VERSION, otherwise this will skip
-                        // entries in the cache preventing those newer
-                        // versions from reading them (that is, until the
-                        // cache is rebuilt).
-                        log::info!("failed to parse {:?} registry package: {}", relative, e);
-                        continue;
-                    }
-                };
-                let version = summary.summary.package_id().version().clone();
-                cache.versions.push((version.clone(), line));
-                ret.versions.insert(version, summary.into());
+        match response {
+            LoadResponse::CacheValid => {
+                log::debug!("fast path for registry cache of {:?}", relative);
+                return Poll::Ready(Ok(cached_summaries));
             }
-            if let Some(index_version) = index_version {
-                cache_bytes = Some(cache.serialize(index_version));
+            LoadResponse::NotFound => {
+                debug_assert!(cached_summaries.is_none());
+                return Poll::Ready(Ok(None));
+            }
+            LoadResponse::Data {
+                raw_data,
+                index_version,
+            } => {
+                // This is the fallback path where we actually talk to the registry backend to load
+                // information. Here we parse every single line in the index (as we need
+                // to find the versions)
+                log::debug!("slow path for {:?}", relative);
+                let mut cache = SummariesCache::default();
+                ret.raw_data = raw_data;
+                for line in split(&ret.raw_data, b'\n') {
+                    // Attempt forwards-compatibility on the index by ignoring
+                    // everything that we ourselves don't understand, that should
+                    // allow future cargo implementations to break the
+                    // interpretation of each line here and older cargo will simply
+                    // ignore the new lines.
+                    let summary = match IndexSummary::parse(config, line, source_id) {
+                        Ok(summary) => summary,
+                        Err(e) => {
+                            // This should only happen when there is an index
+                            // entry from a future version of cargo that this
+                            // version doesn't understand. Hopefully, those future
+                            // versions of cargo correctly set INDEX_V_MAX and
+                            // CURRENT_CACHE_VERSION, otherwise this will skip
+                            // entries in the cache preventing those newer
+                            // versions from reading them (that is, until the
+                            // cache is rebuilt).
+                            log::info!("failed to parse {:?} registry package: {}", relative, e);
+                            continue;
+                        }
+                    };
+                    let version = summary.summary.package_id().version().clone();
+                    cache.versions.push((version.clone(), line));
+                    ret.versions.insert(version, summary.into());
+                }
+                if let Some(index_version) = index_version {
+                    bytes_to_cache = Some(cache.serialize(index_version.as_str()));
+                    version_to_cache = Some(index_version);
+                }
             }
-            Ok(())
-        });
-
-        if result?.is_pending() {
-            assert!(!hit_closure);
-            return Poll::Pending;
-        }
-
-        if !hit_closure {
-            debug_assert!(cache_contents.is_none());
-            return Poll::Ready(Ok(None));
         }
 
         // If we've got debug assertions enabled and the cache was previously
         // present and considered fresh this is where the debug assertions
         // actually happens to verify that our cache is indeed fresh and
         // computes exactly the same value as before.
-        if cfg!(debug_assertions) && cache_contents.is_some() && cache_bytes != cache_contents {
+        let cache_contents = cached_summaries.as_ref().map(|s| &s.raw_data);
+        if cfg!(debug_assertions)
+            && index_version.as_deref() == version_to_cache.as_deref()
+            && cached_summaries.is_some()
+            && bytes_to_cache.as_ref() != cache_contents
+        {
             panic!(
                 "original cache contents:\n{:?}\n\
                  does not equal new cache contents:\n{:?}\n",
                 cache_contents.as_ref().map(|s| String::from_utf8_lossy(s)),
-                cache_bytes.as_ref().map(|s| String::from_utf8_lossy(s)),
+                bytes_to_cache.as_ref().map(|s| String::from_utf8_lossy(s)),
             );
         }
 
@@ -662,7 +673,7 @@ impl Summaries {
         //
         // This is opportunistic so we ignore failure here but are sure to log
         // something in case of error.
-        if let Some(cache_bytes) = cache_bytes {
+        if let Some(cache_bytes) = bytes_to_cache {
             if paths::create_dir_all(cache_path.parent().unwrap()).is_ok() {
                 let path = Filesystem::new(cache_path.clone());
                 config.assert_package_cache_locked(&path);
@@ -677,16 +688,17 @@ impl Summaries {
 
     /// Parses an open `File` which represents information previously cached by
     /// Cargo.
-    pub fn parse_cache(contents: Vec<u8>, last_index_update: &str) -> CargoResult<Summaries> {
-        let cache = SummariesCache::parse(&contents, last_index_update)?;
+    pub fn parse_cache(contents: Vec<u8>) -> CargoResult<(Summaries, InternedString)> {
+        let cache = SummariesCache::parse(&contents)?;
+        let index_version = InternedString::new(cache.index_version);
         let mut ret = Summaries::default();
         for (version, summary) in cache.versions {
             let (start, end) = subslice_bounds(&contents, summary);
             ret.versions
                 .insert(version, MaybeIndexSummary::Unparsed { start, end });
         }
         ret.raw_data = contents;
-        return Ok(ret);
+        return Ok((ret, index_version));
 
         // Returns the start/end offsets of `inner` with `outer`. Asserts that
         // `inner` is a subslice of `outer`.
@@ -742,7 +754,7 @@ impl Summaries {
 const CURRENT_CACHE_VERSION: u8 = 3;
 
 impl<'a> SummariesCache<'a> {
-    fn parse(data: &'a [u8], last_index_update: &str) -> CargoResult<SummariesCache<'a>> {
+    fn parse(data: &'a [u8]) -> CargoResult<SummariesCache<'a>> {
         // NB: keep this method in sync with `serialize` below
         let (first_byte, rest) = data
             .split_first()
@@ -764,18 +776,13 @@ impl<'a> SummariesCache<'a> {
         let rest = &rest[4..];
 
         let mut iter = split(rest, 0);
-        if let Some(update) = iter.next() {
-            if update != last_index_update.as_bytes() {
-                bail!(
-                    "cache out of date: current index ({}) != cache ({})",
-                    last_index_update,
-                    str::from_utf8(update)?,
-                )
-            }
+        let last_index_update = if let Some(update) = iter.next() {
+            str::from_utf8(update)?
         } else {
             bail!("malformed file");
-        }
+        };
         let mut ret = SummariesCache::default();
+        ret.index_version = last_index_update;
         while let Some(version) = iter.next() {
             let version = str::from_utf8(version)?;
             let version = Version::parse(version)?;

src/cargo/sources/registry/local.rs

@@ -1,7 +1,6 @@
 use crate::core::PackageId;
-use crate::sources::registry::{MaybeLock, RegistryConfig, RegistryData};
+use crate::sources::registry::{LoadResponse, MaybeLock, RegistryConfig, RegistryData};
 use crate::util::errors::CargoResult;
-use crate::util::interning::InternedString;
 use crate::util::{Config, Filesystem};
 use cargo_util::{paths, Sha256};
 use std::fs::File;
@@ -48,18 +47,17 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
         path.as_path_unlocked()
     }
 
-    fn current_version(&self) -> Option<InternedString> {
-        None
-    }
-
     fn load(
         &self,
         root: &Path,
         path: &Path,
-        data: &mut dyn FnMut(&[u8]) -> CargoResult<()>,
-    ) -> Poll<CargoResult<()>> {
+        _index_version: Option<&str>,
+    ) -> Poll<CargoResult<LoadResponse>> {
         if self.updated {
-            Poll::Ready(Ok(data(&paths::read_bytes(&root.join(path))?)?))
+            Poll::Ready(Ok(LoadResponse::Data {
+                raw_data: paths::read_bytes(&root.join(path))?,
+                index_version: None,
+            }))
         } else {
             Poll::Pending
         }

src/cargo/sources/registry/mod.rs

@@ -411,6 +411,20 @@ impl<'a> RegistryDependency<'a> {
     }
 }
 
+pub enum LoadResponse {
+    /// The cache is valid. The cached data should be used.
+    CacheValid,
+
+    /// The cache is out of date. Returned data should be used.
+    Data {
+        raw_data: Vec<u8>,
+        index_version: Option<String>,
+    },
+
+    /// The requested crate was found.
+    NotFound,
+}
+
 /// An abstract interface to handle both a local (see `local::LocalRegistry`)
 /// and remote (see `remote::RemoteRegistry`) registry.
 ///
@@ -432,15 +446,13 @@ pub trait RegistryData {
     ///
     /// * `root` is the root path to the index.
     /// * `path` is the relative path to the package to load (like `ca/rg/cargo`).
-    /// * `data` is a callback that will receive the raw bytes of the index JSON file.
-    ///
-    /// If `load` returns a `Poll::Pending` then it must not have called data.
+    /// * `index_version` is the version of the requested crate data currently in cache.
     fn load(
         &self,
         root: &Path,
         path: &Path,
-        data: &mut dyn FnMut(&[u8]) -> CargoResult<()>,
-    ) -> Poll<CargoResult<()>>;
+        index_version: Option<&str>,
+    ) -> Poll<CargoResult<LoadResponse>>;
 
     /// Loads the `config.json` file and returns it.
     ///
@@ -495,15 +507,6 @@ pub trait RegistryData {
     /// Returns the [`Path`] to the [`Filesystem`].
     fn assert_index_locked<'a>(&self, path: &'a Filesystem) -> &'a Path;
 
-    /// Returns the current "version" of the index.
-    ///
-    /// For local registries, this returns `None` because there is no
-    /// versioning. For remote registries, this returns the SHA hash of the
-    /// git index on disk (or None if the index hasn't been downloaded yet).
-    ///
-    /// This is used by index caching to check if the cache is out of date.
-    fn current_version(&self) -> Option<InternedString>;
-
     /// Block until all outstanding Poll::Pending requests are Poll::Ready.
     fn block_until_ready(&mut self) -> CargoResult<()>;
 }