Merge pull request #481 from Kobzol/refresh-ext

Make refresh checks more robust

Author: Kobzol

.sqlx/query-0f0acf10f909aa60beaedde032c1770566948a06c2b2e1f07806e948d83803e1.json

@@ -6,7 +6,6 @@ use chrono::{DateTime, Utc};
 use octocrab::params::checks::CheckRunConclusion;
 use std::collections::BTreeMap;
 
-use crate::bors::PullRequestStatus;
 use crate::bors::RepositoryState;
 use crate::bors::comment::build_timed_out_comment;
 use crate::bors::handlers::workflow::{CancelBuildError, cancel_build};
@@ -177,12 +176,18 @@ pub async fn sync_pull_requests_state(
     // but bors does the merging so it should not happen.
     for pr_num in nonclosed_db_prs_num.keys() {
         if !nonclosed_gh_prs_num.contains_key(pr_num) {
+            let gh_pr = repo
+                .client
+                .get_pull_request(*pr_num)
+                .await
+                .with_context(|| {
+                    anyhow::anyhow!("Cannot fetch PR {}#{pr_num}", repo.repository())
+                })?;
             tracing::debug!(
-                "PR {} not found in open/draft prs in GitHub, closing it in DB",
-                pr_num
+                "PR {pr_num} not found in open/draft prs in GitHub, marking it as {} in DB",
+                gh_pr.status,
             );
-            db.set_pr_status(repo_name, *pr_num, PullRequestStatus::Closed)
-                .await?;
+            db.set_pr_status(repo_name, *pr_num, gh_pr.status).await?;
         }
     }
 

.sqlx/query-3cc3661fb1e1ff945e415f80637e7d7614ffac5358b73ff039220ffed15c4fce.json

@@ -230,7 +230,7 @@ async fn handle_start_auto_build(
             tracing::debug!("Failed to start auto build for PR {pr_num} due to merge conflict");
 
             ctx.db
-                .update_pr_mergeable_state(pr, MergeableState::Unknown)
+                .set_pr_mergeable_state(repo.repository(), pr.number, MergeableState::Unknown)
                 .await?;
             repo.client
                 .post_comment(pr.number, merge_conflict_comment(&gh_pr.head.name))

src/bors/handlers/refresh.rs

@@ -53,7 +53,7 @@ enum QueueMessage {
     Item(MergeabilityQueueItem),
 }
 
-#[derive(PartialEq, Eq)]
+#[derive(PartialEq, Eq, Debug)]
 struct Item {
     /// When to process item (None = immediate is ordered before Some).
     expiration: Option<Instant>,
@@ -152,6 +152,19 @@ impl MergeabilityQueueSender {
     fn insert_item(&self, item: MergeabilityQueueItem, expiration: Option<Instant>) {
         let mut queue = self.inner.queue.lock().unwrap();
 
+        // Make sure that we don't ever put the same pull request twice into the queue
+        // This might seem a bit inefficient, but linearly iterating through e.g. 1000 PRs should
+        // be fine.
+        // We could maybe reset the attempt counter of the PR if it's "refreshed" from the outside,
+        // but that would require using e.g. Cell to mutate the attempt counter through &, which
+        // doesn't seem necessary at the moment.
+        if queue.iter().any(|entry| match &entry.0.inner {
+            QueueMessage::Shutdown => false,
+            QueueMessage::Item(i) => i.pull_request == item.pull_request,
+        }) {
+            return;
+        }
+
         // Notify when:
         // 1. The current item expires sooner than the head of the queue
         let has_earlier_expiration =
@@ -297,19 +310,12 @@ pub async fn check_mergeability(
     }
 
     // We know the mergeability status, so update it in the DB
-    let pr_model = match ctx
-        .db
-        .get_pull_request(&pull_request.repo, pull_request.pr_number)
-        .await?
-    {
-        Some(model) => model,
-        None => {
-            return Err(anyhow::anyhow!("PR not found in database: {pull_request}"));
-        }
-    };
-
     ctx.db
-        .update_pr_mergeable_state(&pr_model, new_mergeable_state.clone().into())
+        .set_pr_mergeable_state(
+            repo_state.repository(),
+            fetched_pr.number,
+            new_mergeable_state.into(),
+        )
         .await?;
 
     Ok(())
@@ -318,11 +324,10 @@ pub async fn check_mergeability(
 #[cfg(test)]
 mod tests {
     use crate::bors::mergeability_queue::{
-        MergeabilityQueueItem, QueuedPullRequest, create_mergeability_queue,
+        BASE_DELAY, MergeabilityQueueItem, QueuedPullRequest, create_mergeability_queue,
     };
     use crate::github::PullRequestNumber;
     use crate::tests::default_repo_name;
-    use std::time::Duration;
 
     #[tokio::test]
     async fn order_by_pr_number() {
@@ -342,12 +347,10 @@ mod tests {
     #[tokio::test]
     async fn immediate_before_delayed() {
         let (tx, rx) = create_mergeability_queue();
-        tx.enqueue_retry_later(item(5, 1));
-        tx.enqueue_pr(default_repo_name(), 1u64.into());
-        tokio::time::sleep(Duration::from_millis(100)).await;
         tx.enqueue_retry_later(item(10, 1));
+        tx.enqueue_pr(default_repo_name(), 2u64.into());
 
-        for expected in [1, 5, 10] {
+        for expected in [2, 10] {
             assert_eq!(
                 rx.dequeue().await.unwrap().0.pull_request.pr_number.0,
                 expected
@@ -356,17 +359,16 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn duplicated_pr() {
+    async fn deduplicate_duplicated_pr() {
         let (tx, rx) = create_mergeability_queue();
-        // Handle duplicated PRs, still keep ordering by time
-        tx.enqueue_retry_later(item(5, 1)); // this attempt will be bumped to 2
+        // Make sure that we don't handle the same PR multiple times
+        tx.enqueue_retry_later(item(5, 1));
+        tx.enqueue_pr(default_repo_name(), 5u64.into());
         tx.enqueue_pr(default_repo_name(), 5u64.into());
 
-        for (pr, attempt) in [(5, 1), (5, 2)] {
-            let item = rx.dequeue().await.unwrap().0;
-            assert_eq!(item.pull_request.pr_number.0, pr);
-            assert_eq!(item.attempt, attempt);
-        }
+        rx.dequeue().await.unwrap();
+        let res = tokio::time::timeout(BASE_DELAY * 2, rx.dequeue()).await;
+        assert!(res.is_err());
     }
 
     fn item(pr_number: u64, attempt: u32) -> MergeabilityQueueItem {

src/bors/merge_queue.rs

@@ -16,9 +16,9 @@ use super::operations::{
     get_nonclosed_pull_requests, get_pending_builds, get_prs_with_unknown_mergeability_state,
     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,
-    record_tagged_bot_comment, set_pr_assignees, set_pr_priority, set_pr_rollup, set_pr_status,
-    unapprove_pull_request, undelegate_pull_request, update_build_check_run_id,
-    update_build_status, update_mergeable_states_by_base_branch, update_pr_mergeability_state,
+    record_tagged_bot_comment, set_pr_assignees, set_pr_mergeability_state, set_pr_priority,
+    set_pr_rollup, set_pr_status, unapprove_pull_request, undelegate_pull_request,
+    update_build_check_run_id, update_build_status, update_mergeable_states_by_base_branch,
     update_pr_try_build_id, update_workflow_status, upsert_pull_request, upsert_repository,
 };
 use super::{ApprovalInfo, DelegatedPermission, MergeableState, RunId, UpsertPullRequestParams};
@@ -69,12 +69,13 @@ impl PgDbClient {
         undelegate_pull_request(&self.pool, pr.id).await
     }
 
-    pub async fn update_pr_mergeable_state(
+    pub async fn set_pr_mergeable_state(
         &self,
-        pr: &PullRequestModel,
+        repo: &GithubRepoName,
+        pr_number: PullRequestNumber,
         mergeable_state: MergeableState,
     ) -> anyhow::Result<()> {
-        update_pr_mergeability_state(&self.pool, pr.id, mergeable_state).await
+        set_pr_mergeability_state(&self.pool, repo, pr_number, mergeable_state).await
     }
 
     pub async fn set_pr_assignees(

src/bors/mergeability_queue.rs

@@ -235,16 +235,18 @@ pub(crate) async fn get_nonclosed_pull_requests(
     .await
 }
 
-pub(crate) async fn update_pr_mergeability_state(
+pub(crate) async fn set_pr_mergeability_state(
     executor: impl PgExecutor<'_>,
-    pr_id: i32,
+    repo: &GithubRepoName,
+    pr_number: PullRequestNumber,
     mergeability_state: MergeableState,
 ) -> anyhow::Result<()> {
     measure_db_query("update_pr_mergeability_state", || async {
         sqlx::query!(
-            "UPDATE pull_request SET mergeable_state = $1 WHERE id = $2",
+            "UPDATE pull_request SET mergeable_state = $3 WHERE repository = $1 AND number = $2",
+            repo as &GithubRepoName,
+            pr_number.0 as i32,
             mergeability_state as _,
-            pr_id
         )
         .execute(executor)
         .await?;