Add sanity check for rollup

src/bors/command/parser.rs

@@ -82,11 +82,9 @@ fn extract_text_from_markdown(text: &str) -> String {
 
     for event in md_parser.into_iter() {
         match event {
-            Event::Text(text) | Event::Html(text) => {
+            Event::Text(text) | Event::Html(text) if stack.is_empty() => {
                 // Only consider commands in raw text outside of wrapping elements
-                if stack.is_empty() {
-                    cleaned_text.push_str(&text);
-                }
+                cleaned_text.push_str(&text);
             }
             Event::SoftBreak | Event::HardBreak | Event::End(TagEnd::Paragraph) => {
                 cleaned_text.push('\n');

src/bors/handlers/mod.rs

@@ -1023,6 +1023,8 @@ pub fn invalidation_comment(
                 ""
             }
         ));
+    } else if is_rollup && outcome.closed {
+        append("This rollup was closed.".to_string());
     }
 
     // Rollup member was invalidated

src/bors/handlers/review.rs

@@ -55,15 +55,15 @@ pub(super) async fn command_approve(
         return Ok(());
     }
 
-    if let Some(rollup_mode) = rollup_mode {
-        let is_rollup = db.is_rollup(pr.db).await?;
-        if is_rollup && rollup_mode != RollupMode::Never {
-            repo_state
-                .client
-                .post_comment(pr.number(), rollup_pr_invalid_rollup_mode_comment(), &db)
-                .await?;
-            return Ok(());
-        }
+    if let Some(rollup_mode) = rollup_mode
+        && rollup_mode != RollupMode::Never
+        && db.is_rollup(pr.db).await?
+    {
+        repo_state
+            .client
+            .post_comment(pr.number(), rollup_pr_invalid_rollup_mode_comment(), &db)
+            .await?;
+        return Ok(());
     }
 
     let (approver, unknown_reviewers) = match approver {

src/bors/merge_queue.rs

@@ -1,11 +1,12 @@
 use anyhow::Context;
 use chrono::{DateTime, Utc};
+use futures::StreamExt;
 use std::future::Future;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 use tracing::Instrument;
 
-use super::{MergeType, bors_commit_author, create_merge_commit_message};
+use super::{Comment, MergeType, bors_commit_author, create_merge_commit_message};
 use crate::bors::build::load_workflow_runs;
 use crate::bors::build::{
     StartBuildCheckRun, StartBuildCommit, StartBuildContext, StartBuildError, StartBuildOutcome,
@@ -15,7 +16,7 @@ use crate::bors::comment::{
     auto_build_push_failed_comment, auto_build_started_comment, auto_build_succeeded_comment,
 };
 use crate::bors::handlers::{
-    InvalidationComment, InvalidationInfo, InvalidationReason, invalidate_pr,
+    InvalidationComment, InvalidationInfo, InvalidationReason, invalidate_pr, unapprove_pr,
 };
 use crate::bors::mergeability_queue::{MergeabilityQueueSender, update_pr_with_known_mergeability};
 use crate::bors::{AUTO_BRANCH_NAME, BuildKind, PullRequestStatus, RepositoryState};
@@ -455,6 +456,39 @@ Actual head SHA: {actual_sha}"#,
             .await?;
             Ok(AutoBuildStartOutcome::ContinueToNextPr)
         }
+        StartAutoBuildError::SanityCheckFailed {
+            error: SanityCheckError::RollupMemberMismatch(mut mismatches),
+            pr: gh_pr,
+        } => {
+            // One or more rollup members have a different commit SHA than what was approved.
+            // This should only happen if we missed a PR push webhook (which would normally unapprove
+            // both the PR and close the rollup). Let's unapprove it here instead to unblock the queue.
+
+            // Note: we don't use invalidate_pr here, because we know that the PR is a rollup,
+            // to have more control over the message.
+            unapprove_pr(repo, &ctx.db, pr, &gh_pr).await?;
+
+            mismatches.sort_by_key(|mismatch| mismatch.member);
+
+            let msg = format!(
+                r#"The following member(s) of this rollup had a different HEAD SHA during a merge attempt than the one from which this rollup was created:
+{}
+
+This rollup has been unapproved."#,
+                mismatches
+                    .into_iter()
+                    .map(|member| format!(
+                        "- #{} (expected SHA: {}, current SHA: {})",
+                        member.member, member.expected, member.actual
+                    ))
+                    .collect::<Vec<String>>()
+                    .join("\n")
+            );
+            repo.client
+                .post_comment(gh_pr.number, Comment::new(msg), &ctx.db)
+                .await?;
+            Ok(AutoBuildStartOutcome::ContinueToNextPr)
+        }
         StartAutoBuildError::GitHubError(error) => {
             tracing::debug!(
                 "Failed to start auto build for PR {pr_num} due to a GitHub error: {error:?}"
@@ -492,13 +526,21 @@ enum SanityCheckError {
         actual: CommitSha,
     },
     /// The pull request was not mergeable.
-    NotMergeable { mergeable_state: MergeableState },
+    NotMergeable {
+        mergeable_state: MergeableState,
+    },
     /// The pull request was not open.
-    WrongStatus { status: PullRequestStatus },
+    WrongStatus {
+        status: PullRequestStatus,
+    },
+    RollupMemberMismatch(Vec<RollupMemberMismatch>),
 }
 
 async fn sanity_check_pr(
+    repo_state: &RepositoryState,
+    db: &PgDbClient,
     gh_pr: &PullRequest,
+    db_pr: &PullRequestModel,
     approval_info: &ApprovalInfo,
 ) -> Result<(), SanityCheckError> {
     if gh_pr.status != PullRequestStatus::Open {
@@ -518,9 +560,75 @@ async fn sanity_check_pr(
             actual: gh_pr.head.sha.clone(),
         });
     }
+
+    let mismatches = sanity_check_rollup(repo_state, db, db_pr).await;
+    if !mismatches.is_empty() {
+        return Err(SanityCheckError::RollupMemberMismatch(mismatches));
+    }
+
     Ok(())
 }
 
+/// A rollup member SHA was `actual` instead of `expected` at the time of the merge of the rollup.
+struct RollupMemberMismatch {
+    member: PullRequestNumber,
+    expected: CommitSha,
+    actual: CommitSha,
+}
+
+/// If `pr` is a rollup, perform a sanity check if its member HEAD SHAs on GitHub
+/// match the state recorded in the database.
+/// Returns a list of mismatches.
+async fn sanity_check_rollup(
+    repo_state: &RepositoryState,
+    db: &PgDbClient,
+    pr: &PullRequestModel,
+) -> Vec<RollupMemberMismatch> {
+    let members = match db.get_rollup_members(pr).await {
+        Ok(members) => members,
+        Err(error) => {
+            tracing::error!(
+                "Cannot perform sanity check to see if PR {pr:?} is a rollup: {error:?}"
+            );
+            return vec![];
+        }
+    };
+
+    // The PR is not a rollup
+    if members.is_empty() {
+        return vec![];
+    }
+
+    // Load rollup members concurrently to make it faster
+    let mut mismatches = vec![];
+    let mut pr_stream = futures::stream::iter(members.into_iter().map(|member| async move {
+        let member_num = member.member;
+        (member, repo_state.client.get_pull_request(member_num).await)
+    }))
+    .buffer_unordered(10);
+    while let Some((member, res)) = pr_stream.next().await {
+        let member_gh = match res {
+            Ok(member_gh) => member_gh,
+            Err(error) => {
+                tracing::error!(
+                    "Cannot fetch member PR {} for a sanity check: {error:?}",
+                    member.member
+                );
+                continue;
+            }
+        };
+        if member_gh.head.sha != member.rolled_up_sha {
+            mismatches.push(RollupMemberMismatch {
+                member: member.member,
+                expected: member.rolled_up_sha,
+                actual: member_gh.head.sha,
+            });
+        }
+    }
+
+    mismatches
+}
+
 /// Starts a new auto build for a pull request.
 async fn start_auto_build(
     repo: &RepositoryState,
@@ -536,7 +644,7 @@ async fn start_auto_build(
         .await
         .map_err(StartAutoBuildError::GitHubError)?;
 
-    sanity_check_pr(&gh_pr, approval_info)
+    sanity_check_pr(repo, &ctx.db, &gh_pr, pr, approval_info)
         .await
         .map_err(|error| StartAutoBuildError::SanityCheckFailed {
             error,

src/bors/mergeability_queue.rs

@@ -306,8 +306,8 @@ impl MergeabilityQueueSender {
             .queues
             .lock()
             .unwrap()
-            .iter()
-            .flat_map(|(_, q)| {
+            .values()
+            .flat_map(|q| {
                 let mut queue = q.clone();
                 let mut items = vec![];
                 while let Some(item) = queue.pop() {
@@ -356,8 +356,8 @@ impl MergeabilityQueueSender {
         };
         let queues = self.inner.queues.lock().unwrap();
         queues
-            .iter()
-            .flat_map(|(_, queue)| queue.iter())
+            .values()
+            .flat_map(|queue| queue.iter())
             .find(|entry| entry.0.entry.pull_request == pr_data)
             .and_then(|entry| entry.0.entry.conflict_source)
     }
@@ -420,8 +420,8 @@ impl MergeabilityQueueSender {
         // but that would require using e.g. Cell to mutate the attempt counter through &, which
         // doesn't seem necessary at the moment.
         if queues
-            .iter()
-            .flat_map(|(_, queue)| queue.iter())
+            .values()
+            .flat_map(|queue| queue.iter())
             .any(|entry| entry.0.entry.pull_request == item.pull_request)
         {
             return;

src/database/client.rs

@@ -3,16 +3,16 @@ use super::operations::{
     delete_tagged_bot_comment, find_build, find_pr_by_build, find_rollups_for_member_pr,
     get_last_n_successful_auto_builds, get_nonclosed_pull_requests, get_pending_builds,
     get_prs_with_stale_mergeability_or_approved, get_pull_request, get_repository,
-    get_repository_by_name, get_tagged_bot_comments, get_workflow_urls_for_build,
-    get_workflows_for_build, insert_repo_if_not_exists, is_rollup, record_tagged_bot_comment,
-    set_pr_assignees, set_pr_mergeability_state, set_pr_priority, set_pr_rollup_mode,
-    set_pr_status, set_stale_mergeability_status_by_base_branch, unapprove_pull_request,
-    undelegate_pull_request, update_build, update_pr_try_build_id, update_workflow_status,
-    upsert_pull_request, upsert_repository,
+    get_repository_by_name, get_rollup_members, get_tagged_bot_comments,
+    get_workflow_urls_for_build, get_workflows_for_build, insert_repo_if_not_exists, is_rollup,
+    record_tagged_bot_comment, set_pr_assignees, set_pr_mergeability_state, set_pr_priority,
+    set_pr_rollup_mode, set_pr_status, set_stale_mergeability_status_by_base_branch,
+    unapprove_pull_request, undelegate_pull_request, update_build, update_pr_try_build_id,
+    update_workflow_status, upsert_pull_request, upsert_repository,
 };
 use super::{
     ApprovalInfo, DelegatedPermission, MergeableState, PrimaryKey, RegisterRollupMemberParams,
-    RunId, UpdateBuildParams, UpsertPullRequestParams,
+    RollupMember, RunId, UpdateBuildParams, UpsertPullRequestParams,
 };
 use std::collections::{HashMap, HashSet};
 
@@ -408,6 +408,15 @@ impl PgDbClient {
         is_rollup(&self.pool, pr.id).await
     }
 
+    /// Returns true if the given PR is a rollup.
+    /// If the returned Vec is empty, the given pull request is not a rollup.
+    pub async fn get_rollup_members(
+        &self,
+        pr: &PullRequestModel,
+    ) -> anyhow::Result<Vec<RollupMember>> {
+        get_rollup_members(&self.pool, pr.id).await
+    }
+
     /// Return all rollups that contain `pr` as a member.
     pub async fn find_rollups_for_member_pr(
         &self,

src/database/mod.rs

@@ -683,6 +683,14 @@ pub struct RegisterRollupMemberParams {
     pub rolled_up_sha: CommitSha,
 }
 
+#[derive(Debug)]
+pub struct RollupMember {
+    /// Pull request number of the member PR.
+    pub member: PullRequestNumber,
+    /// HEAD SHA of the member PR at rollup creation.
+    pub rolled_up_sha: CommitSha,
+}
+
 /// Updates the build table with the given fields.
 /// If `None`, the given column will not be updated.
 /// In other words, none of those columns can be set to `None` after being set at least once.

src/database/operations.rs

@@ -3,7 +3,6 @@ use chrono::Utc;
 use sqlx::postgres::PgExecutor;
 use std::collections::{HashMap, HashSet};
 
-use super::ApprovalStatus;
 use super::Assignees;
 use super::BuildModel;
 use super::CommentModel;
@@ -16,6 +15,7 @@ use super::UpsertPullRequestParams;
 use super::WorkflowStatus;
 use super::WorkflowType;
 use super::{ApprovalInfo, PrimaryKey, UpdateBuildParams};
+use super::{ApprovalStatus, RollupMember};
 use crate::bors::PullRequestStatus;
 use crate::bors::RollupMode;
 use crate::bors::comment::CommentTag;
@@ -1168,6 +1168,33 @@ pub(crate) async fn is_rollup(
     .await
 }
 
+pub(crate) async fn get_rollup_members(
+    executor: impl PgExecutor<'_>,
+    pr_id: PrimaryKey,
+) -> anyhow::Result<Vec<RollupMember>> {
+    measure_db_query("get_rollup_members", || async {
+        let rows = sqlx::query!(
+            r#"
+        SELECT pr.number AS number, rm.rolled_up_sha AS sha
+        FROM rollup_member rm
+        JOIN pull_request AS pr ON pr.id = rm.member
+        WHERE rm.rollup = $1
+            "#,
+            pr_id
+        )
+        .fetch_all(executor)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .map(|row| RollupMember {
+                member: PullRequestNumber(row.number as u64),
+                rolled_up_sha: CommitSha(row.sha),
+            })
+            .collect())
+    })
+    .await
+}
+
 pub(crate) async fn find_rollups_for_member_pr(
     executor: impl PgExecutor<'_>,
     rollup_id: PrimaryKey,