Add merge queue refresh event

Author: Sakib25800

src/bin/bors.rs

@@ -32,6 +32,9 @@ const MERGEABILITY_STATUS_INTERVAL: Duration = Duration::from_secs(60 * 10);
 /// How often should the bot synchronize PR state.
 const PR_STATE_PERIODIC_REFRESH: Duration = Duration::from_secs(60 * 10);
 
+/// How often should the bot process the merge queue.
+const MERGE_QUEUE_INTERVAL: Duration = Duration::from_secs(30);
+
 #[derive(clap::Parser)]
 struct Opts {
     /// Github App ID.
@@ -160,6 +163,7 @@ fn try_main(opts: Opts) -> anyhow::Result<()> {
         let mut cancel_builds_refresh = make_interval(CANCEL_TIMED_OUT_BUILDS_INTERVAL);
         let mut mergeability_status_refresh = make_interval(MERGEABILITY_STATUS_INTERVAL);
         let mut prs_interval = make_interval(PR_STATE_PERIODIC_REFRESH);
+        let mut merge_queue_interval = make_interval(MERGE_QUEUE_INTERVAL);
         loop {
             tokio::select! {
                 _ = config_refresh.tick() => {
@@ -177,6 +181,9 @@ fn try_main(opts: Opts) -> anyhow::Result<()> {
                 _ = prs_interval.tick() => {
                     refresh_tx.send(BorsGlobalEvent::RefreshPullRequestState).await?;
                 }
+                _ = merge_queue_interval.tick() => {
+                    refresh_tx.send(BorsGlobalEvent::ProcessMergeQueue).await?;
+                }
             }
         }
     };

src/bors/event.rs

@@ -75,6 +75,8 @@ pub enum BorsGlobalEvent {
     RefreshPullRequestMergeability,
     /// Periodic event that serves for synchronizing PR state.
     RefreshPullRequestState,
+    /// Process the merge queue.
+    ProcessMergeQueue,
 }
 
 #[derive(Debug)]

src/bors/handlers/mod.rs

@@ -19,6 +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::{BorsContext, Comment, RepositoryState};
 use crate::database::{DelegatedPermission, PullRequestModel};
 use crate::github::{GithubUser, PullRequest, PullRequestNumber};
@@ -238,6 +239,7 @@ pub async fn handle_bors_global_event(
     gh_client: &Octocrab,
     team_api_client: &TeamApiClient,
     mergeable_queue_tx: MergeableQueueSender,
+    merge_queue_tx: MergeQueueSender,
 ) -> anyhow::Result<()> {
     let db = Arc::clone(&ctx.db);
     match event {
@@ -313,6 +315,9 @@ pub async fn handle_bors_global_event(
             #[cfg(test)]
             crate::bors::WAIT_FOR_PR_STATUS_REFRESH.mark();
         }
+        BorsGlobalEvent::ProcessMergeQueue => {
+            merge_queue_tx.send(()).await?;
+        }
     }
     Ok(())
 }

src/bors/merge_queue.rs

@@ -5,15 +5,14 @@ use tracing::Instrument;
 
 use crate::BorsContext;
 
-pub type MergeQueueEvent = ();
+type MergeQueueEvent = ();
+pub type MergeQueueSender = mpsc::Sender<MergeQueueEvent>;
 
 pub async fn merge_queue_tick(_ctx: Arc<BorsContext>) -> anyhow::Result<()> {
     Ok(())
 }
 
-pub fn start_merge_queue(
-    ctx: Arc<BorsContext>,
-) -> (mpsc::Sender<MergeQueueEvent>, impl Future<Output = ()>) {
+pub fn start_merge_queue(ctx: Arc<BorsContext>) -> (MergeQueueSender, impl Future<Output = ()>) {
     let (tx, mut rx) = mpsc::channel::<MergeQueueEvent>(10);
 
     let fut = async move {

src/github/server.rs

@@ -1,5 +1,5 @@
 use crate::bors::event::BorsEvent;
-use crate::bors::merge_queue::{MergeQueueEvent, start_merge_queue};
+use crate::bors::merge_queue::{MergeQueueSender, start_merge_queue};
 use crate::bors::mergeable_queue::{
     MergeableQueueReceiver, MergeableQueueSender, create_mergeable_queue,
     handle_mergeable_queue_item,
@@ -195,7 +195,7 @@ pub async fn github_webhook_handler(
 pub struct BorsProcess {
     pub repository_tx: mpsc::Sender<BorsRepositoryEvent>,
     pub global_tx: mpsc::Sender<BorsGlobalEvent>,
-    pub merge_queue_tx: mpsc::Sender<MergeQueueEvent>,
+    pub merge_queue_tx: MergeQueueSender,
     pub mergeable_queue_tx: MergeableQueueSender,
     pub bors_process: Pin<Box<dyn Future<Output = ()> + Send>>,
 }
@@ -293,7 +293,7 @@ async fn consume_global_events(
     ctx: Arc<BorsContext>,
     mut global_rx: mpsc::Receiver<BorsGlobalEvent>,
     mergeable_queue_tx: MergeableQueueSender,
-    merge_queue_tx: mpsc::Sender<MergeQueueEvent>,
+    merge_queue_tx: MergeQueueSender,
     gh_client: Octocrab,
     team_api: TeamApiClient,
 ) {

src/tests/mocks/bors.rs

@@ -16,7 +16,7 @@ use super::pull_request::{
     GitHubPullRequestEventPayload, GitHubPushEventPayload, PullRequestChangeEvent,
 };
 use super::repository::PullRequest;
-use crate::bors::merge_queue::MergeQueueEvent;
+use crate::bors::merge_queue::MergeQueueSender;
 use crate::bors::mergeable_queue::MergeableQueueSender;
 use crate::bors::{
     RollupMode, WAIT_FOR_CANCEL_TIMED_OUT_BUILDS_REFRESH, WAIT_FOR_MERGEABILITY_STATUS_REFRESH,
@@ -109,7 +109,7 @@ pub struct BorsTester {
     github: GitHubState,
     db: Arc<PgDbClient>,
     mergeable_queue_tx: MergeableQueueSender,
-    merge_queue_tx: Sender<MergeQueueEvent>,
+    merge_queue_tx: MergeQueueSender,
     // Sender for bors global events
     global_tx: Sender<BorsGlobalEvent>,
     // When this field is false, no webhooks should be generated from BorsTester methods