Implement mergeable queue loop

src/bin/bors.rs

@@ -6,8 +6,9 @@ use std::time::Duration;
 
 use anyhow::Context;
 use bors::{
-    BorsContext, BorsGlobalEvent, CommandParser, PgDbClient, ServerState, TeamApiClient,
-    WebhookSecret, create_app, create_bors_process, create_github_client, load_repositories,
+    BorsContext, BorsGlobalEvent, BorsProcess, CommandParser, PgDbClient, ServerState,
+    TeamApiClient, WebhookSecret, create_app, create_bors_process, create_github_client,
+    load_repositories,
 };
 use clap::Parser;
 use sqlx::postgres::PgConnectOptions;
@@ -109,7 +110,12 @@ fn try_main(opts: Opts) -> anyhow::Result<()> {
     }
 
     let ctx = BorsContext::new(CommandParser::new(opts.cmd_prefix), Arc::new(db), repos);
-    let (repository_tx, global_tx, bors_process) = create_bors_process(ctx, client, team_api);
+    let BorsProcess {
+        repository_tx,
+        global_tx,
+        bors_process,
+        ..
+    } = create_bors_process(ctx, client, team_api);
 
     let refresh_tx = global_tx.clone();
     let refresh_process = async move {

src/bors/handlers/mod.rs

@@ -31,6 +31,8 @@ use tracing::Instrument;
 #[cfg(test)]
 use crate::tests::util::TestSyncMarker;
 
+use super::mergeable_queue::MergeableQueueSender;
+
 mod help;
 mod info;
 mod labels;
@@ -48,6 +50,7 @@ pub static WAIT_FOR_WORKFLOW_STARTED: TestSyncMarker = TestSyncMarker::new();
 pub async fn handle_bors_repository_event(
     event: BorsRepositoryEvent,
     ctx: Arc<BorsContext>,
+    mergeable_queue_tx: MergeableQueueSender,
 ) -> anyhow::Result<()> {
     let db = Arc::clone(&ctx.db);
     let Some(repo) = ctx
@@ -129,7 +132,7 @@ pub async fn handle_bors_repository_event(
             let span =
                 tracing::info_span!("Pull request edited", repo = payload.repository.to_string());
 
-            handle_pull_request_edited(repo, db, payload)
+            handle_pull_request_edited(repo, db, mergeable_queue_tx, payload)
                 .instrument(span.clone())
                 .await?;
         }
@@ -199,7 +202,7 @@ pub async fn handle_bors_repository_event(
             let span =
                 tracing::info_span!("Pushed to branch", repo = payload.repository.to_string());
 
-            handle_push_to_branch(repo, db, payload)
+            handle_push_to_branch(repo, db, mergeable_queue_tx, payload)
                 .instrument(span.clone())
                 .await?;
         }

src/bors/handlers/pr_events.rs

@@ -5,6 +5,7 @@ use crate::bors::event::{
     PushToBranch,
 };
 use crate::bors::handlers::labels::handle_label_trigger;
+use crate::bors::mergeable_queue::MergeableQueueSender;
 use crate::bors::{Comment, PullRequestStatus, RepositoryState};
 use crate::database::MergeableState;
 use crate::github::{CommitSha, LabelTrigger, PullRequestNumber};
@@ -13,6 +14,7 @@ use std::sync::Arc;
 pub(super) async fn handle_pull_request_edited(
     repo_state: Arc<RepositoryState>,
     db: Arc<PgDbClient>,
+    mergeable_queue: MergeableQueueSender,
     payload: PullRequestEdited,
 ) -> anyhow::Result<()> {
     let pr = &payload.pull_request;
@@ -32,6 +34,8 @@ pub(super) async fn handle_pull_request_edited(
         return Ok(());
     };
 
+    mergeable_queue.enqueue(pr_model.repository.clone(), pr_number);
+
     if !pr_model.is_approved() {
         return Ok(());
     }
@@ -154,6 +158,7 @@ pub(super) async fn handle_pull_request_ready_for_review(
 pub(super) async fn handle_push_to_branch(
     repo_state: Arc<RepositoryState>,
     db: Arc<PgDbClient>,
+    mergeable_queue: MergeableQueueSender,
     payload: PushToBranch,
 ) -> anyhow::Result<()> {
     let rows = db
@@ -163,8 +168,18 @@ pub(super) async fn handle_push_to_branch(
             MergeableState::Unknown,
         )
         .await?;
+    let affected_prs = db
+        .get_nonclosed_pull_requests_by_base_branch(repo_state.repository(), &payload.branch)
+        .await?;
 
-    tracing::info!("Updated mergeable_state to `unknown` for {} PR(s)", rows);
+    tracing::info!(
+        "Adding {} PR(s) to the mergeable queue due to base branch change",
+        rows
+    );
+
+    for pr in affected_prs {
+        mergeable_queue.enqueue(repo_state.repository().clone(), pr.number);
+    }
 
     Ok(())
 }
@@ -207,7 +222,7 @@ mod tests {
     use crate::bors::PullRequestStatus;
     use crate::tests::mocks::default_pr_number;
     use crate::{
-        database::MergeableState,
+        database::{MergeableState, OctocrabMergeableState},
         tests::mocks::{User, default_branch_name, default_repo_name, run_test},
     };
 
@@ -344,7 +359,7 @@ mod tests {
         run_test(pool.clone(), |mut tester| async {
             tester
                 .edit_pr(default_repo_name(), default_pr_number(), |pr| {
-                    pr.mergeable_state = octocrab::models::pulls::MergeableState::Dirty;
+                    pr.mergeable_state = OctocrabMergeableState::Dirty;
                 })
                 .await?;
             tester
@@ -425,6 +440,7 @@ mod tests {
         .await;
     }
 
+    #[tracing_test::traced_test]
     #[sqlx::test]
     async fn open_and_merge_pr(pool: sqlx::PgPool) {
         run_test(pool, |mut tester| async {
@@ -444,4 +460,51 @@ mod tests {
         })
         .await;
     }
+
+    #[sqlx::test]
+    async fn mergeable_queue_processes_pr_base_change(pool: sqlx::PgPool) {
+        run_test(pool, |mut tester| async {
+            let branch = tester.create_branch("beta").clone();
+            tester
+                .edit_pr(default_repo_name(), default_pr_number(), |pr| {
+                    pr.base_branch = branch;
+                    pr.mergeable_state = OctocrabMergeableState::Unknown;
+                })
+                .await?;
+            tester
+                .wait_for_default_pr(|pr| pr.mergeable_state == MergeableState::Unknown)
+                .await?;
+            tester
+                .default_repo()
+                .lock()
+                .get_pr_mut(default_pr_number())
+                .mergeable_state = OctocrabMergeableState::Dirty;
+            tester
+                .wait_for_default_pr(|pr| pr.mergeable_state == MergeableState::HasConflicts)
+                .await?;
+            Ok(tester)
+        })
+        .await;
+    }
+
+    #[sqlx::test]
+    async fn enqueue_prs_on_push_to_branch(pool: sqlx::PgPool) {
+        run_test(pool, |mut tester| async {
+            tester.open_pr(default_repo_name(), false).await?;
+            tester.push_to_branch(default_branch_name()).await?;
+            tester
+                .wait_for_default_pr(|pr| pr.mergeable_state == MergeableState::Unknown)
+                .await?;
+            tester
+                .default_repo()
+                .lock()
+                .get_pr_mut(default_pr_number())
+                .mergeable_state = OctocrabMergeableState::Dirty;
+            tester
+                .wait_for_default_pr(|pr| pr.mergeable_state == MergeableState::HasConflicts)
+                .await?;
+            Ok(tester)
+        })
+        .await;
+    }
 }