Merge pull request #721 from Mark-Simulacrum/fix-finishers

Fix finishing comments being posted early

Author: Mark-Simulacrum

collector/src/main.rs

@@ -207,14 +207,23 @@ fn bench(
         );
     }
 
-    let interned_cid = rt.block_on(conn.artifact_id(&cid));
-
-    let start = Instant::now();
     let steps = benchmarks
         .iter()
         .map(|b| b.name.to_string())
         .collect::<Vec<_>>();
-    rt.block_on(conn.collector_start(interned_cid, &steps));
+
+    // Make sure there is no observable time when the artifact ID is available
+    // but the in-progress steps are not.
+    let interned_cid = {
+        let mut tx = rt.block_on(conn.transaction());
+        let interned_cid = rt.block_on(tx.conn().artifact_id(&cid));
+        rt.block_on(tx.conn().collector_start(interned_cid, &steps));
+
+        rt.block_on(tx.commit()).unwrap();
+        interned_cid
+    };
+
+    let start = Instant::now();
     let mut skipped = false;
     for (nth_benchmark, benchmark) in benchmarks.iter().enumerate() {
         let is_fresh =

database/src/pool.rs

@@ -71,8 +71,7 @@ pub trait Connection: Send + Sync {
     async fn collector_start_step(&self, aid: ArtifactIdNumber, step: &str) -> bool;
     async fn collector_end_step(&self, aid: ArtifactIdNumber, step: &str);
 
-    // Returns an artifact that is in progress.
-    async fn in_progress_artifact(&self) -> Option<ArtifactId>;
+    async fn in_progress_artifacts(&self) -> Vec<ArtifactId>;
 
     async fn in_progress_steps(&self, aid: &ArtifactId) -> Vec<Step>;
 }

database/src/pool/postgres.rs

@@ -863,7 +863,7 @@ where
             log::error!("did not end {} for {:?}", step, aid);
         }
     }
-    async fn in_progress_artifact(&self) -> Option<ArtifactId> {
+    async fn in_progress_artifacts(&self) -> Vec<ArtifactId> {
         let rows = self
             .conn()
             .query(
@@ -872,34 +872,46 @@ where
             )
             .await
             .unwrap();
-        let aid = rows.into_iter().next().map(|row| row.get::<_, i16>(0))?;
+        let aids = rows
+            .into_iter()
+            .map(|row| row.get::<_, i16>(0))
+            .collect::<Vec<_>>();
 
-        let row = self
-            .conn()
-            .query_one(
-                "select name, date, type from artifact where id = $1",
-                &[&aid],
-            )
-            .await
-            // If we couldn't find the row, early exit with `None`. This is a
-            // transition state -- wouldn't be needed in the long run.
-            .ok()?;
-
-        let ty = row.get::<_, String>(2);
-        Some(match ty.as_str() {
-            "try" | "master" => ArtifactId::Commit(Commit {
-                sha: row.get(0),
-                date: row
-                    .get::<_, Option<_>>(1)
-                    .map(Date)
-                    .unwrap_or_else(|| Date::ymd_hms(2001, 01, 01, 0, 0, 0)),
-            }),
-            "release" => ArtifactId::Artifact(row.get(0)),
-            _ => {
-                log::error!("unknown ty {:?}", ty);
-                return None;
-            }
-        })
+        let mut artifacts = Vec::new();
+        for aid in aids {
+            let row = self
+                .conn()
+                .query_one(
+                    "select name, date, type from artifact where id = $1",
+                    &[&aid],
+                )
+                .await;
+
+            let row = match row {
+                Ok(row) => row,
+                Err(err) => {
+                    log::error!("skipping aid={} -- no such artifact: {:?}", aid, err);
+                    continue;
+                }
+            };
+
+            let ty = row.get::<_, String>(2);
+            artifacts.push(match ty.as_str() {
+                "try" | "master" => ArtifactId::Commit(Commit {
+                    sha: row.get(0),
+                    date: row
+                        .get::<_, Option<_>>(1)
+                        .map(Date)
+                        .unwrap_or_else(|| Date::ymd_hms(2001, 01, 01, 0, 0, 0)),
+                }),
+                "release" => ArtifactId::Artifact(row.get(0)),
+                _ => {
+                    log::error!("unknown ty {:?}", ty);
+                    continue;
+                }
+            });
+        }
+        artifacts
     }
     async fn in_progress_steps(&self, artifact: &ArtifactId) -> Vec<crate::Step> {
         let aid = self.artifact_id(artifact).await;

database/src/pool/sqlite.rs

@@ -675,46 +675,50 @@ impl Connection for SqliteConnection {
             log::error!("did not end {} for {:?}", step, aid);
         }
     }
-    async fn in_progress_artifact(&self) -> Option<ArtifactId> {
-        let aid = self
-            .raw_ref()
-            .query_row(
-                "select distinct aid from collector_progress where end is null order by aid limit 1",
-                params![],
-                |r| r.get::<_, i16>(0),
-            )
-            .optional()
-            .unwrap()?;
-
-        let (name, date, ty) = self
-            .raw_ref()
-            .query_row(
-                "select name, date, type from artifact where id = ?",
-                params![&aid],
-                |r| {
-                    Ok((
-                        r.get::<_, String>(0)?,
-                        r.get::<_, Option<i64>>(1)?,
-                        r.get::<_, String>(2)?,
-                    ))
-                },
+    async fn in_progress_artifacts(&self) -> Vec<ArtifactId> {
+        let conn = self.raw_ref();
+        let mut aids = conn
+            .prepare(
+                "select distinct aid from collector_progress \
+                where end is null order by aid limit 1",
             )
             .unwrap();
 
-        Some(match ty.as_str() {
-            "try" | "master" => ArtifactId::Commit(Commit {
-                sha: name,
-                date: date
-                    .map(|d| Utc.timestamp(d, 0))
-                    .map(Date)
-                    .unwrap_or_else(|| Date::ymd_hms(2001, 01, 01, 0, 0, 0)),
-            }),
-            "release" => ArtifactId::Artifact(name),
-            _ => {
-                log::error!("unknown ty {:?}", ty);
-                return None;
-            }
-        })
+        let aids = aids.query_map(params![], |r| r.get::<_, i16>(0)).unwrap();
+
+        let mut artifacts = Vec::new();
+        for aid in aids {
+            let aid = aid.unwrap();
+            let (name, date, ty) = conn
+                .query_row(
+                    "select name, date, type from artifact where id = ?",
+                    params![&aid],
+                    |r| {
+                        Ok((
+                            r.get::<_, String>(0)?,
+                            r.get::<_, Option<i64>>(1)?,
+                            r.get::<_, String>(2)?,
+                        ))
+                    },
+                )
+                .unwrap();
+
+            artifacts.push(match ty.as_str() {
+                "try" | "master" => ArtifactId::Commit(Commit {
+                    sha: name,
+                    date: date
+                        .map(|d| Utc.timestamp(d, 0))
+                        .map(Date)
+                        .unwrap_or_else(|| Date::ymd_hms(2001, 01, 01, 0, 0, 0)),
+                }),
+                "release" => ArtifactId::Artifact(name),
+                _ => {
+                    log::error!("unknown ty {:?}", ty);
+                    continue;
+                }
+            });
+        }
+        artifacts
     }
     async fn in_progress_steps(&self, artifact: &ArtifactId) -> Vec<crate::Step> {
         let aid = self.artifact_id(artifact).await;

site/src/api.rs

@@ -190,7 +190,7 @@ pub mod status {
 
     #[derive(Serialize, Debug)]
     pub struct Response {
-        pub last_commit: Commit,
+        pub last_commit: Option<Commit>,
         pub benchmarks: Vec<BenchmarkStatus>,
         pub missing: Vec<(Commit, MissingReason)>,
         pub current: Option<CurrentState>,

site/src/github.rs

@@ -582,6 +582,11 @@ where
 }
 
 pub async fn post_finished(data: &InputData) {
+    // If the github token is not configured, do not run this -- we don't want
+    // to mark things as complete without posting the comment.
+    if data.config.keys.github.is_none() {
+        return;
+    }
     let conn = data.conn().await;
     let index = data.index.load();
     let mut commits = index
@@ -591,26 +596,17 @@ pub async fn post_finished(data: &InputData) {
         .collect::<HashSet<_>>();
     let queued = conn.queued_commits().await;
 
-    // In theory this is insufficient -- there could be multiple commits in
-    // progress -- but in practice that really shouldn't happen.
-    //
-    // The failure case here is also just prematurely posting a single comment,
-    // which should be fine. mark_complete below will ensure that only happens
-    // once.
-    //
-    // If this becomes more of a problem, it should be fairly easy to instead
-    // query that there are no in progress benchmarks for commit X.
-    match conn.in_progress_artifact().await {
-        None => {}
-        Some(ArtifactId::Commit(c)) => {
-            commits.insert(c.sha);
-        }
-        Some(ArtifactId::Artifact(_)) => {
-            // do nothing, for now, though eventually we'll want an artifact
-            // queue
+    for aid in conn.in_progress_artifacts().await {
+        match aid {
+            ArtifactId::Commit(c) => {
+                commits.remove(&c.sha);
+            }
+            ArtifactId::Artifact(_) => {
+                // do nothing, for now, though eventually we'll want an artifact
+                // queue
+            }
         }
     }
-
     for commit in queued {
         if !commits.contains(&commit.sha) {
             continue;

site/src/load.rs

@@ -201,23 +201,24 @@ impl InputData {
             .chain(missing)
             .collect::<Vec<_>>();
 
-        match self.conn().await.in_progress_artifact().await {
-            None => {}
-            Some(ArtifactId::Commit(c)) => {
-                commits.insert(
-                    0,
-                    (
-                        rustc_artifacts::Commit {
-                            sha: c.sha,
-                            time: c.date.0,
-                        },
-                        MissingReason::InProgress,
-                    ),
-                );
-            }
-            Some(ArtifactId::Artifact(_)) => {
-                // do nothing, for now, though eventually we'll want an artifact
-                // queue
+        for aid in self.conn().await.in_progress_artifacts().await {
+            match aid {
+                ArtifactId::Commit(c) => {
+                    commits.insert(
+                        0,
+                        (
+                            rustc_artifacts::Commit {
+                                sha: c.sha,
+                                time: c.date.0,
+                            },
+                            MissingReason::InProgress,
+                        ),
+                    );
+                }
+                ArtifactId::Artifact(_) => {
+                    // do nothing, for now, though eventually we'll want an artifact
+                    // queue
+                }
             }
         }
 

site/src/server.rs

@@ -249,35 +249,12 @@ fn prettify_log(log: &str) -> Option<String> {
 
 pub async fn handle_status_page(data: Arc<InputData>) -> status::Response {
     let idx = data.index.load();
-    let last_commit = idx.commits().last().unwrap().clone();
-
-    let mut benchmark_state = data
-        .conn()
-        .await
-        .get_error(ArtifactId::from(last_commit.clone()).lookup(&idx).unwrap())
-        .await
-        .into_iter()
-        .map(|(name, error)| {
-            let msg = if let Some(error) = error {
-                Some(prettify_log(&error).unwrap_or(error))
-            } else {
-                None
-            };
-            status::BenchmarkStatus {
-                name,
-                success: msg.is_none(),
-                error: msg,
-            }
-        })
-        .collect::<Vec<_>>();
-
-    benchmark_state.sort_by_key(|s| s.error.is_some());
-    benchmark_state.reverse();
+    let last_commit = idx.commits().last().cloned();
 
     let missing = data.missing_commits().await;
     // FIXME: no current builds
     let conn = data.conn().await;
-    let current = if let Some(artifact) = conn.in_progress_artifact().await {
+    let current = if let Some(artifact) = conn.in_progress_artifacts().await.pop() {
         let steps = conn
             .in_progress_steps(&artifact)
             .await
@@ -298,6 +275,33 @@ pub async fn handle_status_page(data: Arc<InputData>) -> status::Response {
         None
     };
 
+    let errors = if let Some(last) = &last_commit {
+        data.conn()
+            .await
+            .get_error(ArtifactId::from(last.clone()).lookup(&idx).unwrap())
+            .await
+    } else {
+        Default::default()
+    };
+    let mut benchmark_state = errors
+        .into_iter()
+        .map(|(name, error)| {
+            let msg = if let Some(error) = error {
+                Some(prettify_log(&error).unwrap_or(error))
+            } else {
+                None
+            };
+            status::BenchmarkStatus {
+                name,
+                success: msg.is_none(),
+                error: msg,
+            }
+        })
+        .collect::<Vec<_>>();
+
+    benchmark_state.sort_by_key(|s| s.error.is_some());
+    benchmark_state.reverse();
+
     status::Response {
         last_commit,
         benchmarks: benchmark_state,