Trigger merge queue on events

- PR approval
- Tree state change
- Auto build completion

Author: Sakib25800

src/bors/handlers/mod.rs

@@ -19,7 +19,7 @@ use crate::bors::handlers::trybuild::{TRY_BRANCH_NAME, command_try_build, comman
 use crate::bors::handlers::workflow::{
     handle_check_suite_completed, handle_workflow_completed, handle_workflow_started,
 };
-use crate::bors::merge_queue::MergeQueueSender;
+use crate::bors::merge_queue::{AUTO_BRANCH_NAME, MergeQueueSender};
 use crate::bors::{BorsContext, Comment, RepositoryState};
 use crate::database::{DelegatedPermission, PullRequestModel};
 use crate::github::{GithubUser, PullRequest, PullRequestNumber};
@@ -53,6 +53,7 @@ pub async fn handle_bors_repository_event(
     event: BorsRepositoryEvent,
     ctx: Arc<BorsContext>,
     mergeable_queue_tx: MergeableQueueSender,
+    merge_queue_tx: MergeQueueSender,
 ) -> anyhow::Result<()> {
     let db = Arc::clone(&ctx.db);
     let Some(repo) = ctx
@@ -82,9 +83,10 @@ pub async fn handle_bors_repository_event(
                 author = comment.author.username
             );
             let pr_number = comment.pr_number;
-            if let Err(error) = handle_comment(Arc::clone(&repo), db, ctx, comment)
-                .instrument(span.clone())
-                .await
+            if let Err(error) =
+                handle_comment(Arc::clone(&repo), db, ctx, comment, merge_queue_tx.clone())
+                    .instrument(span.clone())
+                    .await
             {
                 repo.client
                     .post_comment(
@@ -117,7 +119,7 @@ pub async fn handle_bors_repository_event(
                 repo = payload.repository.to_string(),
                 id = payload.run_id.into_inner()
             );
-            handle_workflow_completed(repo, db, payload)
+            handle_workflow_completed(repo, db, payload, &merge_queue_tx)
                 .instrument(span.clone())
                 .await?;
         }
@@ -126,7 +128,7 @@ pub async fn handle_bors_repository_event(
                 "Check suite completed",
                 repo = payload.repository.to_string(),
             );
-            handle_check_suite_completed(repo, db, payload)
+            handle_check_suite_completed(repo, db, payload, &merge_queue_tx)
                 .instrument(span.clone())
                 .await?;
         }
@@ -353,6 +355,7 @@ async fn handle_comment(
     database: Arc<PgDbClient>,
     ctx: Arc<BorsContext>,
     comment: PullRequestComment,
+    merge_queue_tx: MergeQueueSender,
 ) -> anyhow::Result<()> {
     use std::fmt::Write;
 
@@ -404,13 +407,14 @@ async fn handle_comment(
                             &approver,
                             priority,
                             rollup,
+                            &merge_queue_tx,
                         )
                         .instrument(span)
                         .await
                     }
                     BorsCommand::OpenTree => {
                         let span = tracing::info_span!("TreeOpen");
-                        command_open_tree(repo, database, &pr, &comment.author)
+                        command_open_tree(repo, database, &pr, &comment.author, &merge_queue_tx)
                             .instrument(span)
                             .await
                     }
@@ -423,6 +427,7 @@ async fn handle_comment(
                             &comment.author,
                             priority,
                             &comment.html_url,
+                            &merge_queue_tx,
                         )
                         .instrument(span)
                         .await
@@ -561,7 +566,7 @@ async fn reload_repos(
 
 /// Is this branch interesting for the bot?
 fn is_bors_observed_branch(branch: &str) -> bool {
-    branch == TRY_BRANCH_NAME
+    branch == TRY_BRANCH_NAME || branch == AUTO_BRANCH_NAME
 }
 
 /// Deny permission for a request.

src/bors/handlers/review.rs

@@ -8,6 +8,7 @@ use crate::bors::comment::approve_non_open_pr_comment;
 use crate::bors::handlers::has_permission;
 use crate::bors::handlers::labels::handle_label_trigger;
 use crate::bors::handlers::{PullRequestData, deny_request};
+use crate::bors::merge_queue::MergeQueueSender;
 use crate::bors::{Comment, PullRequestStatus};
 use crate::database::ApprovalInfo;
 use crate::database::DelegatedPermission;
@@ -18,6 +19,7 @@ use crate::permissions::PermissionType;
 
 /// Approve a pull request.
 /// A pull request can only be approved by a user of sufficient authority.
+#[allow(clippy::too_many_arguments)]
 pub(super) async fn command_approve(
     repo_state: Arc<RepositoryState>,
     db: Arc<PgDbClient>,
@@ -26,6 +28,7 @@ pub(super) async fn command_approve(
     approver: &Approver,
     priority: Option<u32>,
     rollup: Option<RollupMode>,
+    merge_queue_tx: &MergeQueueSender,
 ) -> anyhow::Result<()> {
     tracing::info!("Approving PR {}", pr.number());
     if !has_permission(&repo_state, author, pr, PermissionType::Review).await? {
@@ -52,6 +55,8 @@ pub(super) async fn command_approve(
 
     db.approve(&pr.db, approval_info, priority, rollup).await?;
     handle_label_trigger(&repo_state, pr.number(), LabelTrigger::Approved).await?;
+
+    merge_queue_tx.send(()).await?;
     notify_of_approval(&repo_state, pr, approver.as_str()).await
 }
 
@@ -156,6 +161,7 @@ pub(super) async fn command_close_tree(
     author: &GithubUser,
     priority: u32,
     comment_url: &str,
+    merge_queue_tx: &MergeQueueSender,
 ) -> anyhow::Result<()> {
     if !sufficient_approve_permission(repo_state.clone(), author) {
         deny_request(&repo_state, pr.number(), author, PermissionType::Review).await?;
@@ -169,6 +175,8 @@ pub(super) async fn command_close_tree(
         },
     )
     .await?;
+
+    merge_queue_tx.send(()).await?;
     notify_of_tree_closed(&repo_state, pr.number(), priority).await
 }
 
@@ -177,6 +185,7 @@ pub(super) async fn command_open_tree(
     db: Arc<PgDbClient>,
     pr: &PullRequestData,
     author: &GithubUser,
+    merge_queue_tx: &MergeQueueSender,
 ) -> anyhow::Result<()> {
     if !sufficient_delegate_permission(repo_state.clone(), author) {
         deny_request(&repo_state, pr.number(), author, PermissionType::Review).await?;
@@ -185,6 +194,8 @@ pub(super) async fn command_open_tree(
 
     db.upsert_repository(repo_state.repository(), TreeState::Open)
         .await?;
+
+    merge_queue_tx.send(()).await?;
     notify_of_tree_open(&repo_state, pr.number()).await
 }
 

src/bors/handlers/workflow.rs

@@ -11,6 +11,8 @@ use crate::bors::comment::{try_build_succeeded_comment, workflow_failed_comment}
 use crate::bors::event::{CheckSuiteCompleted, WorkflowCompleted, WorkflowStarted};
 use crate::bors::handlers::is_bors_observed_branch;
 use crate::bors::handlers::labels::handle_label_trigger;
+use crate::bors::merge_queue::AUTO_BRANCH_NAME;
+use crate::bors::merge_queue::MergeQueueSender;
 use crate::database::{BuildStatus, WorkflowStatus};
 use crate::github::LabelTrigger;
 
@@ -66,6 +68,7 @@ pub(super) async fn handle_workflow_completed(
     repo: Arc<RepositoryState>,
     db: Arc<PgDbClient>,
     mut payload: WorkflowCompleted,
+    merge_queue_tx: &MergeQueueSender,
 ) -> anyhow::Result<()> {
     if !is_bors_observed_branch(&payload.branch) {
         return Ok(());
@@ -98,13 +101,14 @@ pub(super) async fn handle_workflow_completed(
         branch: payload.branch,
         commit_sha: payload.commit_sha,
     };
-    try_complete_build(repo.as_ref(), db.as_ref(), event).await
+    try_complete_build(repo.as_ref(), db.as_ref(), event, merge_queue_tx).await
 }
 
 pub(super) async fn handle_check_suite_completed(
     repo: Arc<RepositoryState>,
     db: Arc<PgDbClient>,
     payload: CheckSuiteCompleted,
+    merge_queue_tx: &MergeQueueSender,
 ) -> anyhow::Result<()> {
     if !is_bors_observed_branch(&payload.branch) {
         return Ok(());
@@ -115,14 +119,15 @@ pub(super) async fn handle_check_suite_completed(
         payload.branch,
         payload.commit_sha
     );
-    try_complete_build(repo.as_ref(), db.as_ref(), payload).await
+    try_complete_build(repo.as_ref(), db.as_ref(), payload, merge_queue_tx).await
 }
 
 /// Try to complete a pending build.
 async fn try_complete_build(
     repo: &RepositoryState,
     db: &PgDbClient,
     payload: CheckSuiteCompleted,
+    merge_queue_tx: &MergeQueueSender,
 ) -> anyhow::Result<()> {
     if !is_bors_observed_branch(&payload.branch) {
         return Ok(());
@@ -217,6 +222,11 @@ async fn try_complete_build(
     };
     repo.client.post_comment(pr.number, message).await?;
 
+    // Trigger merge queue when an auto build completes
+    if payload.branch == AUTO_BRANCH_NAME {
+        merge_queue_tx.send(()).await?;
+    }
+
     Ok(())
 }
 

src/bors/merge_queue.rs

@@ -7,8 +7,33 @@ use crate::BorsContext;
 use crate::bors::RepositoryState;
 use crate::utils::sort_queue::sort_queue_prs;
 
-type MergeQueueEvent = ();
-pub type MergeQueueSender = mpsc::Sender<MergeQueueEvent>;
+#[derive(Debug)]
+enum MergeQueueEvent {
+    Trigger,
+    Shutdown,
+}
+
+#[derive(Clone)]
+pub struct MergeQueueSender {
+    inner: mpsc::Sender<MergeQueueEvent>,
+}
+
+impl MergeQueueSender {
+    pub async fn send(&self, _value: ()) -> Result<(), mpsc::error::SendError<()>> {
+        self.inner
+            .send(MergeQueueEvent::Trigger)
+            .await
+            .map_err(|_| mpsc::error::SendError(()))
+    }
+
+    pub fn shutdown(&self) {
+        let _ = self.inner.try_send(MergeQueueEvent::Shutdown);
+    }
+}
+
+/// Branch where CI checks run for auto builds.
+/// This branch should run CI checks.
+pub(super) const AUTO_BRANCH_NAME: &str = "automation/bors/auto";
 
 pub async fn merge_queue_tick(ctx: Arc<BorsContext>) -> anyhow::Result<()> {
     let repos: Vec<Arc<RepositoryState>> =
@@ -46,25 +71,35 @@ pub async fn merge_queue_tick(ctx: Arc<BorsContext>) -> anyhow::Result<()> {
 
 pub fn start_merge_queue(ctx: Arc<BorsContext>) -> (MergeQueueSender, impl Future<Output = ()>) {
     let (tx, mut rx) = mpsc::channel::<MergeQueueEvent>(10);
+    let sender = MergeQueueSender { inner: tx };
 
     let fut = async move {
-        while rx.recv().await.is_some() {
-            let span = tracing::info_span!("MergeQueue");
-            tracing::debug!("Processing merge queue");
-            if let Err(error) = merge_queue_tick(ctx.clone()).instrument(span.clone()).await {
-                // In tests, we want to panic on all errors.
-                #[cfg(test)]
-                {
-                    panic!("Merge queue handler failed: {error:?}");
+        while let Some(event) = rx.recv().await {
+            match event {
+                MergeQueueEvent::Trigger => {
+                    let span = tracing::info_span!("MergeQueue");
+                    tracing::debug!("Processing merge queue");
+                    if let Err(error) = merge_queue_tick(ctx.clone()).instrument(span.clone()).await
+                    {
+                        // In tests, we want to panic on all errors.
+                        #[cfg(test)]
+                        {
+                            panic!("Merge queue handler failed: {error:?}");
+                        }
+                        #[cfg(not(test))]
+                        {
+                            use crate::utils::logging::LogError;
+                            span.log_error(error);
+                        }
+                    }
                 }
-                #[cfg(not(test))]
-                {
-                    use crate::utils::logging::LogError;
-                    span.log_error(error);
+                MergeQueueEvent::Shutdown => {
+                    tracing::debug!("Merge queue received shutdown signal");
+                    break;
                 }
             }
         }
     };
 
-    (tx, fut)
+    (sender, fut)
 }

src/github/server.rs

@@ -216,6 +216,7 @@ pub fn create_bors_process(
 
     let (merge_queue_tx, merge_queue_fut) = start_merge_queue(ctx.clone());
     let merge_queue_tx_clone = merge_queue_tx.clone();
+    let merge_queue_tx_for_return = merge_queue_tx.clone();
 
     let service = async move {
         // In tests, we shutdown these futures by dropping the channel sender,
@@ -225,7 +226,12 @@ pub fn create_bors_process(
         #[cfg(test)]
         {
             let _ = tokio::join!(
-                consume_repository_events(ctx.clone(), repository_rx, mq_tx.clone()),
+                consume_repository_events(
+                    ctx.clone(),
+                    repository_rx,
+                    mq_tx.clone(),
+                    merge_queue_tx_clone.clone()
+                ),
                 consume_global_events(
                     ctx.clone(),
                     global_rx,
@@ -244,7 +250,7 @@ pub fn create_bors_process(
         #[cfg(not(test))]
         {
             tokio::select! {
-                _ = consume_repository_events(ctx.clone(), repository_rx, mq_tx.clone()) => {
+                _ = consume_repository_events(ctx.clone(), repository_rx, mq_tx.clone(), merge_queue_tx_clone.clone()) => {
                     tracing::error!("Repository event handling process has ended");
                 }
                 _ = consume_global_events(ctx.clone(), global_rx, mq_tx, merge_queue_tx, gh_client, team_api) => {
@@ -263,8 +269,8 @@ pub fn create_bors_process(
     BorsProcess {
         repository_tx,
         global_tx,
-        merge_queue_tx: merge_queue_tx_clone,
         mergeable_queue_tx,
+        merge_queue_tx: merge_queue_tx_for_return,
         bors_process: Box::pin(service),
     }
 }
@@ -273,16 +279,18 @@ async fn consume_repository_events(
     ctx: Arc<BorsContext>,
     mut repository_rx: mpsc::Receiver<BorsRepositoryEvent>,
     mergeable_queue_tx: MergeableQueueSender,
+    merge_queue_tx: MergeQueueSender,
 ) {
     while let Some(event) = repository_rx.recv().await {
         let ctx = ctx.clone();
         let mergeable_queue_tx = mergeable_queue_tx.clone();
 
         let span = tracing::info_span!("RepositoryEvent");
         tracing::debug!("Received repository event: {event:?}");
-        if let Err(error) = handle_bors_repository_event(event, ctx, mergeable_queue_tx)
-            .instrument(span.clone())
-            .await
+        if let Err(error) =
+            handle_bors_repository_event(event, ctx, mergeable_queue_tx, merge_queue_tx.clone())
+                .instrument(span.clone())
+                .await
         {
             handle_root_error(span, error);
         }

src/tests/mocks/bors.rs

@@ -839,7 +839,7 @@ impl BorsTester {
         // Make sure that the event channel senders are closed
         drop(self.app);
         drop(self.global_tx);
-        drop(self.merge_queue_tx);
+        self.merge_queue_tx.shutdown();
         self.mergeable_queue_tx.shutdown();
         // Wait until all events are handled in the bors service
         bors.await.unwrap();