pr conflicts: bound the forge list and skip branches no pr targets

swarm-controller/src/pr_conflicts.rs

@@ -4,7 +4,8 @@
 //! Forgejo sends no delivery when a push to a branch changes whether the pull
 //! requests targeting it merge, and the router holds no forge client. So the
 //! controller watches the pushes it already receives on the `vcs-activity`
-//! hook: after a push to a branch it waits for the forge to recompute, lists
+//! hook: after a push to the default branch, or to a branch it has seen an
+//! open pull request target ([`OpenPulls`]), it waits for the forge to recompute, lists
 //! the open pull requests that target the branch, and publishes one
 //! [`MERGEABILITY_EVENT`] per pull request. The router keeps whether each pull
 //! request was already conflicting and acts only on a change.
@@ -41,6 +42,8 @@ pub(crate) struct BranchPush {
     pub(crate) repo: String,
     pub(crate) repo_id: i64,
     pub(crate) branch: String,
+    /// The repo's default branch, empty when the body names none.
+    pub(crate) default_branch: String,
 }
 
 impl BranchPush {
@@ -61,6 +64,11 @@ impl BranchPush {
             repo: repository.get("full_name")?.as_str()?.to_owned(),
             repo_id: repository.get("id")?.as_i64()?,
             branch: branch.to_owned(),
+            default_branch: repository
+                .get("default_branch")
+                .and_then(Json::as_str)
+                .unwrap_or_default()
+                .to_owned(),
         })
     }
 
@@ -90,6 +98,15 @@ impl Debounce {
         true
     }
 
+    /// Free `key` whatever pushes arrived, for a check that ended without
+    /// finishing.
+    pub(crate) fn release(&self, key: &(String, String)) {
+        self.running
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .remove(key);
+    }
+
     /// A check of `key` is done. True when a push arrived meanwhile, so the
     /// check runs again; false when it was the last, and the branch is free.
     pub(crate) fn finish(&self, key: &(String, String)) -> bool {
@@ -223,21 +240,134 @@ where
     }
 }
 
+/// The open PRs the controller has seen on a `pull_request` delivery, by
+/// (repo, number), with the branch each targets.
+///
+/// Only a push to the repo's default branch, or to a branch in here, starts a
+/// check; a push to any other branch has no PR to find. The set lives in
+/// memory and is empty after a restart, so a PR that targets a branch other
+/// than the default (a stacked PR) is checked again only from its next
+/// `pull_request` delivery.
+#[derive(Default)]
+pub(crate) struct OpenPulls {
+    bases: Mutex<HashMap<(String, i64), String>>,
+}
+
+impl OpenPulls {
+    /// Take in a `pull_request` family delivery: an open PR is recorded under
+    /// its target branch (replacing an earlier one if it was retargeted), a
+    /// closed one is forgotten. A body that is no PR delivery changes nothing.
+    pub(crate) fn note(&self, body: &[u8]) {
+        let Ok(json) = serde_json::from_slice::<Json>(body) else {
+            return;
+        };
+        let (Some(pull), Some(repo)) = (
+            json.get("pull_request"),
+            json.pointer("/repository/full_name").and_then(Json::as_str),
+        ) else {
+            return;
+        };
+        let Some(number) = pull.get("number").and_then(Json::as_i64) else {
+            return;
+        };
+        let key = (repo.to_owned(), number);
+        let closed = json.get("action").and_then(Json::as_str) == Some("closed")
+            || pull.get("state").and_then(Json::as_str) == Some("closed");
+        let mut bases = self.bases.lock().unwrap_or_else(PoisonError::into_inner);
+        match pull.pointer("/base/ref").and_then(Json::as_str) {
+            Some(base) if !closed => {
+                bases.insert(key, base.to_owned());
+            }
+            _ => {
+                bases.remove(&key);
+            }
+        }
+    }
+
+    fn targets(&self, repo: &str, branch: &str) -> bool {
+        self.bases
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .iter()
+            .any(|((r, _), base)| r == repo && base == branch)
+    }
+}
+
+static OPEN_PULLS: LazyLock<OpenPulls> = LazyLock::new(OpenPulls::default);
+
+/// A `pull_request` family delivery arrived.
+pub(crate) fn on_pull_request(body: &[u8]) {
+    OPEN_PULLS.note(body);
+}
+
+/// Whether `push` starts a check: the branch is one a PR can target as far as
+/// the controller knows, and no check of it is running (if one is, it runs
+/// again when done). True means the caller runs the check and frees the
+/// branch afterwards.
+fn begin_check(open: &OpenPulls, running: &Debounce, push: &BranchPush) -> bool {
+    let default = !push.default_branch.is_empty() && push.default_branch == push.branch;
+    (default || open.targets(&push.repo, &push.branch)) && running.begin(push.key())
+}
+
+/// Keeps a branch's key in [`Debounce`] set while a check runs, and frees it
+/// however the check ends, a panic or an abandoned task included.
+struct Held<'a> {
+    debounce: &'a Debounce,
+    key: (String, String),
+    freed: bool,
+}
+
+impl Held<'_> {
+    /// The check is done: true when pushes arrived meanwhile and it runs
+    /// again, false when the branch is free.
+    fn again(&mut self) -> bool {
+        let again = self.debounce.finish(&self.key);
+        self.freed = !again;
+        again
+    }
+}
+
+impl Drop for Held<'_> {
+    fn drop(&mut self) {
+        if !self.freed {
+            self.debounce.release(&self.key);
+        }
+    }
+}
+
+/// `fut`, the forge call named `work`, or an error when the forge does not
+/// answer within the forge deadline. The forge client has no timeout of its
+/// own.
+async fn bounded<T>(work: &str, fut: impl Future<Output = anyhow::Result<T>>) -> anyhow::Result<T> {
+    crate::forge::within_forge_deadline(work, fut)
+        .await
+        .unwrap_or_else(|| {
+            anyhow::bail!(
+                "the forge did not answer within {}s",
+                crate::forge::FORGE_DEADLINE.as_secs()
+            )
+        })
+}
+
 /// A push arrived on the `vcs-activity` hook. Starts the check for its branch
-/// unless one is already under way, in which case that one runs again when it
-/// is done.
+/// unless the branch has no PR to find or a check is already under way, in
+/// which case that one runs again when it is done.
 pub(crate) fn on_push(state: &AppState, body: &[u8]) {
     let Some(push) = BranchPush::parse(body) else {
         return;
     };
-    if RUNNING.begin(push.key()) {
+    if begin_check(&OPEN_PULLS, &RUNNING, &push) {
         tokio::spawn(watch(state.clone(), push));
     }
 }
 
 async fn watch(state: AppState, push: BranchPush) {
+    let mut running = Held {
+        debounce: &RUNNING,
+        key: push.key(),
+        freed: false,
+    };
     let Some((owner, name)) = push.repo.split_once('/') else {
-        RUNNING.finish(&push.key());
         return;
     };
     loop {
@@ -247,7 +377,11 @@ async fn watch(state: AppState, push: BranchPush) {
                 &push,
                 RECHECK,
                 || async {
-                    let listed = forge.open_pulls_on(owner, name, &push.branch).await?;
+                    let listed = bounded(
+                        "list open pulls",
+                        forge.open_pulls_on(owner, name, &push.branch),
+                    )
+                    .await?;
                     Ok(listed.iter().filter_map(Pull::from_listed).collect())
                 },
                 |action, body| {
@@ -264,7 +398,7 @@ async fn watch(state: AppState, push: BranchPush) {
             )
             .await;
         }
-        if !RUNNING.finish(&push.key()) {
+        if !running.again() {
             return;
         }
     }
@@ -280,6 +414,7 @@ mod tests {
             repo: "o/r".to_owned(),
             repo_id: 7,
             branch: "main".to_owned(),
+            default_branch: "main".to_owned(),
         }
     }
 
@@ -474,7 +609,7 @@ mod tests {
         serde_json::to_vec(&json!({
             "ref": git_ref,
             "after": after,
-            "repository": { "id": 7, "full_name": "o/r" },
+            "repository": { "id": 7, "full_name": "o/r", "default_branch": "main" },
         }))
         .expect("json")
     }
@@ -487,6 +622,7 @@ mod tests {
                 repo: "o/r".to_owned(),
                 repo_id: 7,
                 branch: "feat/x".to_owned(),
+                default_branch: "main".to_owned(),
             })
         );
         assert_eq!(
@@ -514,4 +650,113 @@ mod tests {
         assert!(d.begin(key("main")), "so the next push starts a new check");
         assert!(!d.finish(&key("other")));
     }
+
+    fn pull_delivery(action: &str, repo: &str, number: i64, base: &str) -> Vec<u8> {
+        serde_json::to_vec(&json!({
+            "action": action,
+            "pull_request": { "number": number, "base": { "ref": base } },
+            "repository": { "full_name": repo },
+        }))
+        .expect("json")
+    }
+
+    fn on_branch(branch: &str) -> BranchPush {
+        BranchPush {
+            branch: branch.to_owned(),
+            ..push()
+        }
+    }
+
+    /// A push to a branch no PR is known to target starts no check, and
+    /// leaves nothing marked running.
+    #[test]
+    fn a_push_to_a_branch_no_pull_request_targets_starts_nothing() {
+        let (open, running) = (OpenPulls::default(), Debounce::default());
+        let topic = on_branch("agent/topic");
+        assert!(!begin_check(&open, &running, &topic));
+        assert!(running.begin(topic.key()), "nothing was marked running");
+        running.release(&topic.key());
+
+        // Controls: the default branch, and a branch a PR is known to target.
+        assert!(begin_check(&open, &running, &push()));
+        open.note(&pull_delivery("opened", "o/r", 3, "agent/topic"));
+        assert!(begin_check(&open, &running, &topic));
+        assert!(!begin_check(&open, &running, &topic), "one check at a time");
+
+        // A body naming no default branch checks only a branch a PR targets.
+        let unnamed = BranchPush {
+            default_branch: String::new(),
+            ..on_branch("main")
+        };
+        assert!(!begin_check(&open, &running, &unnamed));
+        // Another repo's PR on the same branch name is not this repo's.
+        let other = BranchPush {
+            repo: "o/other".to_owned(),
+            ..topic
+        };
+        assert!(!begin_check(&open, &running, &other));
+    }
+
+    #[test]
+    fn the_known_targets_follow_the_pull_request_deliveries() {
+        let open = OpenPulls::default();
+        open.note(&pull_delivery("opened", "o/r", 5, "base-a"));
+        open.note(&pull_delivery("opened", "o/r", 6, "base-a"));
+        assert!(open.targets("o/r", "base-a"));
+
+        open.note(&pull_delivery("edited", "o/r", 5, "base-b"));
+        assert!(open.targets("o/r", "base-b"), "5 was retargeted");
+        assert!(open.targets("o/r", "base-a"), "6 still targets it");
+
+        open.note(&pull_delivery("closed", "o/r", 6, "base-a"));
+        assert!(!open.targets("o/r", "base-a"));
+        open.note(&pull_delivery("closed", "o/r", 5, "base-b"));
+        assert!(!open.targets("o/r", "base-b"));
+
+        open.note(b"not json");
+        open.note(br#"{"issue":{"number":1}}"#);
+        assert!(!open.targets("o/r", "base-a"), "no PR delivery, no change");
+    }
+
+    /// A forge that never answers ends the check at the deadline and frees the
+    /// branch for the next push.
+    #[tokio::test(start_paused = true)]
+    async fn a_list_that_never_answers_ends_the_check_and_frees_the_branch() {
+        let running = Debounce::default();
+        let p = push();
+        assert!(running.begin(p.key()));
+        let mut guard = Held {
+            debounce: &running,
+            key: p.key(),
+            freed: false,
+        };
+        let start = tokio::time::Instant::now();
+        check(
+            &p,
+            Duration::ZERO,
+            || bounded("list open pulls", std::future::pending()),
+            |_, _| async {},
+        )
+        .await;
+        assert_eq!(start.elapsed(), crate::forge::FORGE_DEADLINE);
+        assert!(!guard.again());
+        assert!(running.begin(p.key()), "the branch is free");
+    }
+
+    #[test]
+    fn a_check_that_panics_frees_the_branch() {
+        let running = Debounce::default();
+        let key = push().key();
+        assert!(running.begin(key.clone()));
+        let ended = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _guard = Held {
+                debounce: &running,
+                key: key.clone(),
+                freed: false,
+            };
+            panic!("the check died");
+        }));
+        assert!(ended.is_err());
+        assert!(running.begin(key), "the branch is free");
+    }
 }

swarm-controller/src/webhook.rs

@@ -407,7 +407,8 @@ pub(super) enum DeliveryKind {
     /// one is instance-wide (registered via `admin_create_hook`, not
     /// `repo_create_hook`) — different scope, same event name.
     ///
-    /// A push to a branch also starts the check of the PRs that target it
+    /// A push to the default branch, or to a branch a PR targets, also starts
+    /// the check of the PRs that target it
     /// (`crate::pr_conflicts`).
     #[strum(serialize = "vcs-activity")]
     VcsActivity,
@@ -657,6 +658,9 @@ pub(super) async fn post_webhook_forge(
         }
         DeliveryKind::ReviewRequest => crate::review_classifier::review::on_delivery(&state, &body),
         DeliveryKind::Issues | DeliveryKind::PullRequest | DeliveryKind::WorkflowRun => {
+            if kind == DeliveryKind::PullRequest {
+                crate::pr_conflicts::on_pull_request(&body);
+            }
             if let Err(not_kept) = crate::forge_events::publish(&state, &headers, body).await {
                 tracing::warn!(kind = <&str>::from(kind), reason = ?not_kept, "webhook: forge event not kept");
                 return (StatusCode::SERVICE_UNAVAILABLE, "forge event not kept").into_response();