rewrite consistency check & make it execute fixes

Author: syphar

Cargo.lock

@@ -16,7 +16,7 @@ exclude = [
 ]
 
 [features]
-consistency_check = ["crates-index", "rayon"]
+consistency_check = ["crates-index", "rayon", "itertools"]
 
 [dependencies]
 sentry = "0.29.0"
@@ -67,6 +67,7 @@ zip = {version = "0.6.2", default-features = false, features = ["bzip2"]}
 bzip2 = "0.4.2"
 serde_cbor = "0.11.1"
 getrandom = "0.2.1"
+itertools = { version = "0.10.5", optional = true}
 
 # Async
 tokio = { version = "1.0", features = ["rt-multi-thread", "signal", "macros"] }

Cargo.toml

@@ -469,7 +469,7 @@ impl DatabaseSubcommand {
 
             #[cfg(feature = "consistency_check")]
             Self::Synchronize { dry_run } => {
-                docs_rs::utils::consistency::run_check(&mut *ctx.conn()?, &*ctx.index()?, dry_run)?;
+                docs_rs::utils::consistency::run_check(&ctx, dry_run)?;
             }
         }
         Ok(())

src/bin/cratesfyi.rs

@@ -266,7 +266,7 @@ fn retry<T>(mut f: impl FnMut() -> Result<T>, max_attempts: u32) -> Result<T> {
 }
 
 #[instrument(skip(conn))]
-fn set_yanked(conn: &mut postgres::Client, name: &str, version: &str, yanked: bool) {
+pub fn set_yanked(conn: &mut postgres::Client, name: &str, version: &str, yanked: bool) {
     let activity = if yanked { "yanked" } else { "unyanked" };
 
     match conn

src/build_queue.rs

@@ -11,7 +11,7 @@ pub use self::pool::{Pool, PoolClient, PoolError};
 
 mod add_package;
 pub mod blacklist;
-mod delete;
+pub mod delete;
 pub(crate) mod file;
 mod migrate;
 mod pool;

src/db/mod.rs

@@ -89,12 +89,7 @@ impl Index {
 
     #[cfg(feature = "consistency_check")]
     pub(crate) fn crates(&self) -> Result<crates_index::Index> {
-        use tracing::debug;
-        // First ensure the index is up to date, peeking will pull the latest changes without
-        // affecting anything else.
-        debug!("Updating index");
-        self.diff()?.peek_changes()?;
-        debug!("Opening with `crates_index`");
+        tracing::debug!("Opening with `crates_index`");
         // crates_index requires the repo url to match the existing origin or it tries to reinitialize the repo
         let repo_url = self
             .repository_url

src/index/mod.rs

@@ -1,48 +1,15 @@
-use std::{
-    cmp::PartialEq,
-    collections::BTreeMap,
-    fmt::{self, Debug, Display, Formatter},
-};
-
-#[derive(Default, Debug)]
-pub(crate) struct Data {
-    pub(crate) crates: BTreeMap<CrateName, Crate>,
-}
-
-#[derive(PartialOrd, Ord, PartialEq, Eq, Clone, Default, Debug)]
-pub(crate) struct CrateName(pub(crate) String);
-
-#[derive(Default, Debug)]
-pub(crate) struct Crate {
-    pub(crate) releases: BTreeMap<Version, Release>,
-}
-
-#[derive(PartialOrd, Ord, PartialEq, Eq, Clone, Default, Debug)]
-pub(crate) struct Version(pub(crate) String);
-
-#[derive(Default, Debug)]
-pub(crate) struct Release {}
-
-impl PartialEq<String> for CrateName {
-    fn eq(&self, other: &String) -> bool {
-        self.0 == *other
-    }
+#[derive(Clone, PartialEq, Debug)]
+pub(super) struct Crate {
+    pub(super) name: String,
+    pub(super) releases: Releases,
 }
 
-impl PartialEq<String> for Version {
-    fn eq(&self, other: &String) -> bool {
-        self.0 == *other
-    }
-}
+pub(super) type Crates = Vec<Crate>;
 
-impl Display for CrateName {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        Display::fmt(&self.0, f)
-    }
-}
+pub(super) type Releases = Vec<Release>;
 
-impl Display for Version {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        Display::fmt(&self.0, f)
-    }
+#[derive(Clone, Debug, PartialEq)]
+pub(super) struct Release {
+    pub(super) version: String,
+    pub(super) yanked: Option<bool>,
 }

src/utils/consistency/data.rs

@@ -1,60 +1,108 @@
-use super::data::{Crate, CrateName, Data, Release, Version};
-use std::collections::BTreeMap;
-
-pub(crate) fn load(conn: &mut postgres::Client) -> Result<Data, anyhow::Error> {
-    let rows = conn.query(
-        "
-        SELECT
-            crates.name,
-            releases.version
-        FROM crates
-        INNER JOIN releases ON releases.crate_id = crates.id
-        ORDER BY crates.id, releases.id
-    ",
-        &[],
+use super::data::{Crate, Crates, Release, Releases};
+use crate::Config;
+use anyhow::Result;
+use itertools::Itertools;
+use postgres::fallible_iterator::FallibleIterator;
+use std::iter;
+
+pub(super) fn load(conn: &mut postgres::Client, config: &Config) -> Result<Crates> {
+    let rows = conn.query_raw(
+        "SELECT name, version, yanked
+         FROM (
+             SELECT
+                 crates.name,
+                 releases.version,
+                 releases.yanked
+             FROM crates
+             INNER JOIN releases ON releases.crate_id = crates.id
+             UNION ALL 
+             -- crates & releases that are already queued 
+             -- don't have to be requeued.
+             SELECT queue.name, queue.version, NULL as yanked
+             FROM queue 
+             LEFT OUTER JOIN crates ON crates.name = queue.name 
+             LEFT OUTER JOIN releases ON (
+                 releases.crate_id = crates.id AND 
+                 releases.version = queue.version
+             )
+             WHERE queue.attempt < $1 AND (
+                 crates.id IS NULL OR 
+                 releases.id IS NULL
+             )
+         ) AS inp 
+         ORDER BY name, version",
+        iter::once(config.build_attempts as i32),
     )?;
 
-    let mut data = Data {
-        crates: BTreeMap::new(),
-    };
+    let mut crates = Crates::new();
 
-    let mut rows = rows.iter();
+    for (crate_name, release_rows) in &rows
+        // `rows` is a `FallibleIterator` which needs to be converted before
+        // we can use Itertools.group_by on it.
+        .iterator()
+        .map(|row| row.expect("error fetching rows"))
+        .into_iter()
+        .group_by(|row| row.get("name"))
+    {
+        let releases: Releases = release_rows
+            .map(|row| Release {
+                version: row.get("version"),
+                yanked: row.get("yanked"),
+            })
+            .collect();
 
-    struct Current {
-        name: CrateName,
-        krate: Crate,
+        crates.push(Crate {
+            name: crate_name,
+            releases,
+        });
     }
 
-    let mut current = if let Some(row) = rows.next() {
-        Current {
-            name: CrateName(row.get("name")),
-            krate: Crate {
-                releases: {
-                    let mut releases = BTreeMap::new();
-                    releases.insert(Version(row.get("version")), Release {});
-                    releases
-                },
-            },
-        }
-    } else {
-        return Ok(data);
-    };
-
-    for row in rows {
-        let name = row.get("name");
-        if current.name != name {
-            data.crates.insert(
-                std::mem::replace(&mut current.name, CrateName(name)),
-                std::mem::take(&mut current.krate),
-            );
-        }
-        current
-            .krate
-            .releases
-            .insert(Version(row.get("version")), Release::default());
-    }
+    Ok(crates)
+}
 
-    data.crates.insert(current.name, current.krate);
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use super::{Crate, Release};
+    use crate::test::wrapper;
 
-    Ok(data)
+    #[test]
+    fn test_load() {
+        wrapper(|env| {
+            env.build_queue().add_crate("queued", "0.0.1", 0, None)?;
+            env.fake_release().name("krate").version("0.0.2").create()?;
+            env.fake_release()
+                .name("krate")
+                .version("0.0.3")
+                .yanked(true)
+                .create()?;
+
+            assert_eq!(
+                load(&mut env.db().conn(), &env.config())?,
+                vec![
+                    Crate {
+                        name: "krate".into(),
+                        releases: vec![
+                            Release {
+                                version: "0.0.2".into(),
+                                yanked: Some(false),
+                            },
+                            Release {
+                                version: "0.0.3".into(),
+                                yanked: Some(true),
+                            }
+                        ]
+                    },
+                    Crate {
+                        name: "queued".into(),
+                        releases: vec![Release {
+                            version: "0.0.1".into(),
+                            yanked: None,
+                        }]
+                    },
+                ]
+            );
+            Ok(())
+        })
+    }
 }