Ingest new commits on push

Author: Mark-Simulacrum

Cargo.lock

@@ -35,6 +35,7 @@ rustc-artifacts = "0.2"
 itertools = "0.9"
 hashbrown = "0.7"
 rocksdb = "0.14"
+arc-swap = "0.4"
 
 [dependencies.collector]
 path = "../collector"

site/Cargo.toml

@@ -1,48 +1,6 @@
-use anyhow::Context as _;
-use rocksdb::WriteBatch;
-use std::fs;
-use std::io::Read as _;
 use std::path::Path;
 
-use collector::{ArtifactData, CommitData};
-use site::db::{CollectionId, DbLabel, Label, LabelPath, LabelTag, Profile};
-
-enum Res {
-    Artifact(ArtifactData),
-    Commit(CommitData),
-}
-
-fn deserialize_path(path: &Path) -> Res {
-    let mut file = fs::File::open(path)
-        .with_context(|| format!("Failed to open {}", path.display()))
-        .unwrap();
-    let mut file_contents = Vec::new();
-    if path.extension().map_or(false, |e| e == "sz") {
-        let mut szip_reader = snap::read::FrameDecoder::new(std::io::BufReader::new(file));
-        szip_reader
-            .read_to_end(&mut file_contents)
-            .with_context(|| format!("Failed to read {}", path.display()))
-            .unwrap();
-    } else {
-        file.read_to_end(&mut file_contents)
-            .with_context(|| format!("Failed to read {}", path.display()))
-            .unwrap();
-    };
-
-    if path
-        .file_name()
-        .unwrap()
-        .to_str()
-        .unwrap()
-        .starts_with("artifact-")
-    {
-        Res::Artifact(serde_json::from_slice(&file_contents).unwrap())
-    } else {
-        Res::Commit(serde_json::from_slice(&file_contents).unwrap())
-    }
-}
-
-fn main() -> Result<(), Box<dyn std::error::Error>> {
+fn main() {
     let dir = std::env::args().nth(1).expect("database as first arg");
     let db = site::db::open(&dir, true);
     let mut index = site::db::Index::load(&db);
@@ -53,90 +11,17 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     for (idx, path) in paths.into_iter().enumerate() {
         if idx % 10 == 0 {
             eprintln!(
-                "{}/{}, per {:?}; total {:?}",
+                "{}/{}, per {:?}; estimated total time {:?}",
                 idx,
                 paths_count,
                 last.elapsed() / 10,
                 last.elapsed() / 10 * paths_count as u32
             );
             last = std::time::Instant::now();
         }
-        let res = deserialize_path(std::path::Path::new(&path));
-        let (cid, benchmarks) = match res {
-            Res::Commit(cd) => (CollectionId::Commit(cd.commit), cd.benchmarks),
-            Res::Artifact(ad) => (CollectionId::Artifact(ad.id), ad.benchmarks),
-        };
-
-        let cid_num = index.intern_cid(&cid);
-        let mut path = LabelPath::new();
-
-        for (name, bres) in benchmarks.into_iter() {
-            path.set(Label::Crate(name));
-            let benchmark = match bres {
-                Ok(b) => b,
-                Err(e) => {
-                    let mut batch = WriteBatch::default();
-                    index.insert_labeled(&DbLabel::Errors { krate: name }, &mut batch, cid_num, &e);
-                    db.write(batch).unwrap();
-                    path.remove(LabelTag::Crate);
-                    continue;
-                }
-            };
-
-            for run in &benchmark.runs {
-                let mut batch = WriteBatch::default();
-                let profile = if run.check {
-                    Profile::Check
-                } else if run.release {
-                    Profile::Opt
-                } else {
-                    Profile::Debug
-                };
-                let state = (&run.state).into();
-                path.set(Label::Profile(profile));
-                path.set(Label::Cache(state));
-
-                for (sid, stat) in run.stats.iter() {
-                    path.set(Label::ProcessStat(sid.as_pstat()));
-                    index.insert_labeled(
-                        &DbLabel::ProcessStat {
-                            krate: name,
-                            profile,
-                            cache: state,
-                            stat: sid.as_pstat(),
-                        },
-                        &mut batch,
-                        cid_num,
-                        &stat,
-                    );
-                }
-                path.remove(LabelTag::ProcessStat);
-
-                if let Some(self_profile) = &run.self_profile {
-                    for qd in self_profile.query_data.iter() {
-                        path.set(Label::Query(qd.label));
-                        index.insert_labeled(
-                            &DbLabel::SelfProfileQuery {
-                                krate: name,
-                                profile,
-                                cache: state,
-                                query: qd.label,
-                            },
-                            &mut batch,
-                            cid_num,
-                            &site::db::QueryDatum::from_query_data(qd),
-                        );
-                    }
-                    path.remove(LabelTag::Query);
-                }
-                db.write_without_wal(batch).unwrap();
-            }
-        }
+        site::ingest::ingest(&db, &mut index, Path::new(&path));
     }
 
     index.store(&db);
     db.flush().unwrap();
-    std::mem::drop(db);
-
-    Ok(())
 }

site/src/bin/ingest.rs

@@ -103,23 +103,25 @@ pub fn data_for(data: &[Commit], is_left: bool, query: Bound) -> Option<Commit>
     }
 }
 
-pub fn range_subset(data: &[Commit], range: RangeInclusive<Bound>) -> &[Commit] {
+pub fn range_subset(data: Vec<Commit>, range: RangeInclusive<Bound>) -> Vec<Commit> {
     let (a, b) = range.into_inner();
 
     let left_idx = data.iter().position(|commit| a.left_match(commit));
     let right_idx = data.iter().rposition(|commit| b.left_match(commit));
 
     if let (Some(left), Some(right)) = (left_idx, right_idx) {
-        data.get(left..=right).unwrap_or_else(|| {
-            log::error!(
-                "Failed to compute left/right indices from {:?}..={:?}",
-                a,
-                b
-            );
-            &[]
-        })
+        data.get(left..=right)
+            .map(|s| s.to_vec())
+            .unwrap_or_else(|| {
+                log::error!(
+                    "Failed to compute left/right indices from {:?}..={:?}",
+                    a,
+                    b
+                );
+                vec![]
+            })
     } else {
-        &[]
+        vec![]
     }
 }
 
@@ -567,7 +569,7 @@ pub fn open(at: &str, ingest: bool) -> DB {
     DB::open_cf_descriptors(&db_opts, &at, descriptors).unwrap()
 }
 
-#[derive(Deserialize, Serialize, Default)]
+#[derive(Clone, Deserialize, Serialize, Default)]
 pub struct Index {
     commits: Indexed<Commit>,
     artifacts: Indexed<Box<str>>,
@@ -577,7 +579,7 @@ pub struct Index {
     queries: Indexed<(Crate, Profile, Cache, QueryLabel)>,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Clone, Serialize, Deserialize)]
 struct Indexed<T> {
     #[serde(with = "index_serde")]
     #[serde(bound = "T: Serialize + serde::de::DeserializeOwned + std::hash::Hash + Eq")]
@@ -753,7 +755,9 @@ impl Index {
     }
 
     pub fn commits(&self) -> Vec<Commit> {
-        self.commits.map.keys().copied().collect()
+        let mut commits = self.commits.map.keys().copied().collect::<Vec<_>>();
+        commits.sort();
+        commits
     }
 
     // FIXME: in theory this won't scale indefinitely as there's potentially

site/src/db.rs

@@ -8,40 +8,58 @@
 // except according to those terms.
 
 use std::path::Path;
+use std::path::PathBuf;
 use std::process::Command;
 
 use crate::CommandFailed;
 
-const BRANCH: &'static str = "master";
-const GIT: &'static str = "git";
-const GIT_CHECKOUT: &'static str = "checkout";
-const GIT_PULL: &'static str = "pull";
-
-pub fn update_repo(repo_path: &str) -> anyhow::Result<()> {
+pub fn update_repo(repo_path: &str) -> anyhow::Result<Vec<PathBuf>> {
     let working_dir = Path::new(repo_path);
-    checkout_master(working_dir)?;
-    pull_updates(working_dir)?;
-    Ok(())
+    execute_command(working_dir, &["checkout", "master"])?;
+    let original = get_commit_hash(working_dir)?;
+    execute_command(working_dir, &["pull"])?;
+
+    let output = Command::new("git")
+        .current_dir(working_dir)
+        .arg("diff")
+        .arg("--name-only")
+        .arg(&original)
+        .output()?;
+    if !output.status.success() {
+        Err(CommandFailed {
+            command: format!("git diff --name-only {}", original),
+        })?
+    }
+    Ok(String::from_utf8(output.stdout)?
+        .lines()
+        .map(|l| PathBuf::from(repo_path).join(l.trim()))
+        .collect())
+}
+
+fn get_commit_hash(working_dir: &Path) -> anyhow::Result<String> {
+    let output = Command::new("git")
+        .current_dir(working_dir)
+        .arg("rev-parse")
+        .arg("master")
+        .output()?;
+    if !output.status.success() {
+        Err(CommandFailed {
+            command: format!("git rev-parse master"),
+        })?
+    }
+    Ok(String::from_utf8(output.stdout)?.trim().to_string())
 }
 
 pub fn execute_command(working_dir: &Path, args: &[&str]) -> anyhow::Result<()> {
-    let status = Command::new(GIT)
+    let status = Command::new("git")
         .current_dir(working_dir)
         .args(args)
         .status()?;
     if status.success() {
         Ok(())
     } else {
         Err(CommandFailed {
-            command: format!("{} {:?}", GIT, args),
+            command: format!("git {:?}", args),
         })?
     }
 }
-
-fn checkout_master(working_dir: &Path) -> anyhow::Result<()> {
-    execute_command(working_dir, &[GIT_CHECKOUT, BRANCH])
-}
-
-fn pull_updates(working_dir: &Path) -> anyhow::Result<()> {
-    execute_command(working_dir, &[GIT_PULL])
-}