From 02f21940a458898cc1d2863271144499eb24e413 Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 11 Nov 2026 05:40:28 +0100 Subject: [PATCH] pr conflicts: bound the forge list and skip branches no pr targets --- docs/4-implementation/swarm/controller.md | 2 +- docs/4-implementation/swarm/notify-router.md | 16 +- swarm-controller/src/pr_conflicts.rs | 261 ++++++++++++++++++- swarm-controller/src/webhook.rs | 6 +- 4 files changed, 272 insertions(+), 13 deletions(-) diff --git a/docs/4-implementation/swarm/controller.md b/docs/4-implementation/swarm/controller.md index 654918c39..59b1c4686 100644 --- a/docs/4-implementation/swarm/controller.md +++ b/docs/4-implementation/swarm/controller.md @@ -385,7 +385,7 @@ itself, all under `https:///webhook/forge/`: [notification router](notify-router.md#the-path) The instance-wide `push` hook also starts a check of the open PRs targeting -the pushed branch, which publishes a `pull_request_mergeability` event for each +the pushed branch (the default branch, or a branch the controller has seen a PR target), which publishes a `pull_request_mergeability` event for each to the same stream: [merge conflicts as a source](notify-router.md#flows). The controller **interprets** a delivery and sends hives a specific diff --git a/docs/4-implementation/swarm/notify-router.md b/docs/4-implementation/swarm/notify-router.md index f6c5801cd..4cf1d8dba 100644 --- a/docs/4-implementation/swarm/notify-router.md +++ b/docs/4-implementation/swarm/notify-router.md @@ -218,20 +218,30 @@ it. With no head kept yet, it can't tell and sends the todo. **Merge conflicts as a source.** Forgejo sends no delivery when a push to a branch changes whether the PRs targeting it merge, and the router has no forge client. swarm-controller makes the event instead: on each `push` delivery on its -instance-wide `push` hook, for a branch push that didn't delete the branch, it -waits 30 seconds for the forge to recompute, then lists the open PRs that +instance-wide `push` hook, for a push to the repo's default branch or to a branch the +controller has seen an open PR target, it waits 30 seconds for the forge to +recompute, then lists the open PRs that target the branch (one call per page of 50) and publishes one `pull_request_mergeability` event per PR to `$SWARM.events.forge`, where `action` is `conflicting` or `mergeable`. The body is a `pull_request` delivery, so `repo`, `number`, `title`, `url`, `author` and `assignees` read as for any PR. +The controller learns which branches PRs target from the `pull_request` +deliveries it already handles, in memory: an opened, edited or reopened PR adds +its base, a closed one drops it. A push to any other branch, such as an agent's +topic branch, starts nothing. After a controller restart the set is empty, so the +controller checks a PR that targets a branch other than the default, such as a +stacked PR, again only from its next `pull_request` delivery. A forge call that +gets no answer within two minutes ends the check with a log line, and the next +push starts a new one. + Forgejo's `mergeable` flag is false for a conflict, for an error, for a draft and while the check still runs. The controller leaves drafts out and counts a PR as conflicting only when two reads 30 seconds apart both say false. A PR that reads undecided gets read again, up to four reads in all. One check runs per branch at a time; a push during it makes it run once more when it ends. A -push to a branch with no open PR costs one list call. With open PRs that all +push to the default branch costs one list call when no PR targets it. With open PRs that all merge it's still one; each PR that reads false or undecided adds a second read at least, and four is the most. diff --git a/swarm-controller/src/pr_conflicts.rs b/swarm-controller/src/pr_conflicts.rs index 1d3769153..7c4eaab9e 100644 --- a/swarm-controller/src/pr_conflicts.rs +++ b/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>, +} + +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::(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 = 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(work: &str, fut: impl Future>) -> anyhow::Result { + 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 { + 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"); + } } diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index 4aee907a5..c5f328a77 100644 --- a/swarm-controller/src/webhook.rs +++ b/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();