Sort and retrieve PRs in merge queue

Author: Sakib25800

.sqlx/query-b1fe02180d44f3cdc04ca0be76163606a84a5ed390079e4786e471b2e8850e49.json

@@ -4,11 +4,38 @@ use tokio::sync::mpsc;
 use tracing::Instrument;
 
 use crate::BorsContext;
+use crate::bors::RepositoryState;
+use crate::utils::sort_queue::sort_queue_prs;
 
 type MergeQueueEvent = ();
 pub type MergeQueueSender = mpsc::Sender<MergeQueueEvent>;
 
-pub async fn merge_queue_tick(_ctx: Arc<BorsContext>) -> anyhow::Result<()> {
+pub async fn merge_queue_tick(ctx: Arc<BorsContext>) -> anyhow::Result<()> {
+    let repos: Vec<Arc<RepositoryState>> =
+        ctx.repositories.read().unwrap().values().cloned().collect();
+
+    for repo in repos {
+        let repo_name = repo.repository();
+        let repo_db = match ctx.db.repo_db(repo_name).await? {
+            Some(repo) => repo,
+            None => {
+                tracing::error!("Repository {repo_name} not found");
+                continue;
+            }
+        };
+        let priority = repo_db.tree_state.priority();
+        let prs = ctx.db.get_merge_queue_prs(repo_name, priority).await?;
+
+        // Sort PRs according to merge queue priority rules.
+        // Successful builds come first so they can be merged immediately,
+        // then pending builds (which block the queue to prevent starting simultaneous auto-builds).
+        let prs = sort_queue_prs(prs);
+
+        for _ in prs {
+            // Process PRs...
+        }
+    }
+
     Ok(())
 }
 

src/bors/merge_queue.rs

@@ -1,6 +1,7 @@
 use sqlx::PgPool;
 
 use crate::bors::{PullRequestStatus, RollupMode};
+use crate::database::operations::get_merge_queue_prs;
 use crate::database::{
     BuildModel, BuildStatus, PullRequestModel, RepoModel, TreeState, WorkflowModel, WorkflowStatus,
     WorkflowType,
@@ -308,4 +309,12 @@ impl PgDbClient {
     pub async fn delete_auto_build(&self, pr: &PullRequestModel) -> anyhow::Result<()> {
         delete_auto_build(&self.pool, pr.id).await
     }
+
+    pub async fn get_merge_queue_prs(
+        &self,
+        repo: &GithubRepoName,
+        tree_priority: Option<u32>,
+    ) -> anyhow::Result<Vec<PullRequestModel>> {
+        get_merge_queue_prs(&self.pool, repo, tree_priority.map(|p| p as i32)).await
+    }
 }

src/database/client.rs

@@ -991,3 +991,58 @@ pub(crate) async fn delete_auto_build(
     })
     .await
 }
+
+/// Fetches pull requests eligible for merge:
+/// - Only approved PRs that are open and mergeable
+/// - Includes only PRs with pending or successful auto builds
+/// - Excludes PRs that do not meet the tree closure priority threshold (if tree closed)
+pub(crate) async fn get_merge_queue_prs(
+    executor: impl PgExecutor<'_>,
+    repo: &GithubRepoName,
+    tree_priority: Option<i32>,
+) -> anyhow::Result<Vec<PullRequestModel>> {
+    measure_db_query("get_merge_queue_prs", || async {
+        let records = sqlx::query_as!(
+            PullRequestModel,
+            r#"
+            SELECT
+                pr.id,
+                pr.repository as "repository: GithubRepoName",
+                pr.number as "number!: i64",
+                pr.title,
+                pr.author,
+                pr.assignees as "assignees: Assignees",
+                (
+                    pr.approved_by,
+                    pr.approved_sha
+                ) AS "approval_status!: ApprovalStatus",
+                pr.status as "pr_status: PullRequestStatus",
+                pr.priority,
+                pr.rollup as "rollup: RollupMode",
+                pr.delegated_permission as "delegated_permission: DelegatedPermission",
+                pr.base_branch,
+                pr.mergeable_state as "mergeable_state: MergeableState",
+                pr.created_at as "created_at: DateTime<Utc>",
+                try_build AS "try_build: BuildModel",
+                auto_build AS "auto_build: BuildModel"
+            FROM pull_request as pr
+            LEFT JOIN build AS try_build ON pr.try_build_id = try_build.id
+            LEFT JOIN build AS auto_build ON pr.auto_build_id = auto_build.id
+            WHERE pr.repository = $1
+              AND pr.status = 'open'
+              AND pr.approved_by IS NOT NULL
+              AND pr.mergeable_state = 'mergeable'
+              -- Include only PRs with pending or successful auto builds
+              AND (auto_build.status IS NULL OR auto_build.status IN ('pending', 'success'))
+              -- Tree closure check (if tree_priority is set)
+              AND ($2::int IS NULL OR pr.priority >= $2)
+            "#,
+            repo as &GithubRepoName,
+            tree_priority
+        )
+        .fetch_all(executor)
+        .await?;
+        Ok(records)
+    })
+    .await
+}

src/database/operations.rs

@@ -1,3 +1,4 @@
 pub mod logging;
+pub mod sort_queue;
 pub mod text;
 pub mod timing;

src/utils/mod.rs

@@ -0,0 +1,80 @@
+use crate::bors::RollupMode;
+use crate::database::{BuildStatus, MergeableState, PullRequestModel};
+
+/// Sorts pull requests according to merge queue priority rules.
+/// Ordered by pending builds > success builds > approval > mergeability > priority value > rollup > age.
+pub fn sort_queue_prs(mut prs: Vec<PullRequestModel>) -> Vec<PullRequestModel> {
+    prs.sort_by(|a, b| {
+        // 1. Pending builds come first (to block merge queue)
+        get_queue_blocking_priority(a)
+            .cmp(&get_queue_blocking_priority(b))
+            // 2. Compare approval status (approved PRs should come first)
+            .then_with(|| a.is_approved().cmp(&b.is_approved()).reverse())
+            // 3. Compare build status within approval groups
+            .then_with(|| get_status_priority(a).cmp(&get_status_priority(b)))
+            // 4. Compare mergeable state (0 = mergeable, 1 = conflicts/unknown)
+            .then_with(|| get_mergeable_priority(a).cmp(&get_mergeable_priority(b)))
+            // 5. Compare priority numbers (higher priority should come first)
+            .then_with(|| {
+                a.priority
+                    .unwrap_or(0)
+                    .cmp(&b.priority.unwrap_or(0))
+                    .reverse()
+            })
+            // 6. Compare rollup mode (-1 = never/iffy, 0 = maybe, 1 = always)
+            .then_with(|| {
+                get_rollup_priority(a.rollup.as_ref()).cmp(&get_rollup_priority(b.rollup.as_ref()))
+            })
+            // 7. Compare PR numbers (older first)
+            .then_with(|| a.number.cmp(&b.number))
+    });
+    prs
+}
+
+fn get_queue_blocking_priority(pr: &PullRequestModel) -> u32 {
+    match &pr.auto_build {
+        Some(build) => match build.status {
+            // Pending builds must come first to block the merge queue
+            BuildStatus::Pending => 0,
+            // All other statuses come after
+            _ => 1,
+        },
+        None => 1, // No build - can potentially start new build
+    }
+}
+
+fn get_status_priority(pr: &PullRequestModel) -> u32 {
+    match &pr.auto_build {
+        Some(build) => match build.status {
+            BuildStatus::Success => 1,
+            BuildStatus::Pending => 1, // Same as success since queue blocking is handled separately
+            BuildStatus::Failure => 3,
+            BuildStatus::Cancelled | BuildStatus::Timeouted => 2,
+        },
+        None => {
+            if pr.is_approved() {
+                1 // Approved but no build - should be prioritized
+            } else {
+                2 // No status
+            }
+        }
+    }
+}
+
+fn get_mergeable_priority(pr: &PullRequestModel) -> u32 {
+    match pr.mergeable_state {
+        MergeableState::Mergeable => 0,
+        MergeableState::HasConflicts => 1,
+        MergeableState::Unknown => 1,
+    }
+}
+
+fn get_rollup_priority(rollup: Option<&RollupMode>) -> u32 {
+    match rollup {
+        Some(RollupMode::Always) => 3,
+        Some(RollupMode::Maybe) => 2,
+        None => 2, // Default case
+        Some(RollupMode::Iffy) => 1,
+        Some(RollupMode::Never) => 0,
+    }
+}