//! Telling the notification router when an open pull request stops or starts //! merging into its target branch. //! //! 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 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. //! //! The forge's `mergeable` flag is false for a conflict, for an error, for a //! draft, and while the check still runs. A draft is left out, and a pull //! request counts as conflicting only when two reads one [`RECHECK`] apart both //! say false, which a check still running does not survive. use std::collections::{BTreeSet, HashMap, HashSet}; use std::future::Future; use std::sync::{LazyLock, Mutex, PoisonError}; use std::time::Duration; use forgejo_api::structs::PullRequest; use serde_json::{Value as Json, json}; use swarm_wire::forge_events::MERGEABILITY_EVENT; use crate::routes::AppState; /// How long after a push the forge gets to recompute before the first read. const SETTLE: Duration = Duration::from_secs(30); /// The wait between reads of a pull request that did not read as mergeable. const RECHECK: Duration = Duration::from_secs(30); /// The most reads one check makes. const MAX_READS: usize = 4; /// One push to a branch, as far as this module needs it. #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct BranchPush { /// `owner/name`. 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 { /// The push a `push` webhook body describes, or `None` for anything else: /// a tag, a push that deleted its branch, or a body that is no push. pub(crate) fn parse(body: &[u8]) -> Option { let json: Json = serde_json::from_slice(body).ok()?; let branch = json.get("ref")?.as_str()?.strip_prefix("refs/heads/")?; let deleted = json .get("after") .and_then(Json::as_str) .is_some_and(|after| !after.is_empty() && after.bytes().all(|b| b == b'0')); let repository = json.get("repository")?; if branch.is_empty() || deleted { return None; } Some(Self { 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(), }) } fn key(&self) -> (String, String) { (self.repo.clone(), self.branch.clone()) } } /// The branches with a check under way, so a burst of pushes to one branch is /// one check and not one each. #[derive(Default)] pub(crate) struct Debounce { /// Whether a push arrived since the check last started. running: Mutex>, } impl Debounce { /// A push to `key`. True when no check was running, which the caller /// starts; false when one was, which will run again once it finishes. pub(crate) fn begin(&self, key: (String, String)) -> bool { let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner); if let Some(pushed_again) = running.get_mut(&key) { *pushed_again = true; return false; } running.insert(key, false); 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 { let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner); match running.get_mut(key) { Some(pushed_again) if *pushed_again => { *pushed_again = false; true } _ => { running.remove(key); false } } } } static RUNNING: LazyLock = LazyLock::new(Debounce::default); /// What a check needs of one listed pull request. #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct Pull { pub(crate) number: i64, title: String, url: String, author: String, assignees: BTreeSet, head: String, draft: bool, /// The forge's flag: `None` when it did not say. mergeable: Option, } impl Pull { /// `pull` as listed, or `None` when it names no number or no author, so /// there is nobody to tell. pub(crate) fn from_listed(pull: &PullRequest) -> Option { let login = |u: &forgejo_api::structs::User| u.login.clone(); Some(Self { number: pull.number?, title: pull.title.clone().unwrap_or_default(), url: pull .html_url .as_ref() .map(ToString::to_string) .unwrap_or_default(), author: pull.user.as_ref().and_then(login)?, assignees: pull.assignees.iter().flatten().filter_map(login).collect(), head: pull .head .as_ref() .and_then(|h| h.sha.clone()) .unwrap_or_default(), draft: pull.draft.unwrap_or(false), mergeable: pull.mergeable, }) } /// The body of the event that says `action` (`conflicting` or /// `mergeable`) of this pull request: a `pull_request` delivery the /// router's parser reads. fn event_body(&self, push: &BranchPush, action: &str) -> Vec { let user = |login: &str| json!({ "login": login }); let body = json!({ "action": action, "number": self.number, "pull_request": { "number": self.number, "title": self.title, "html_url": self.url, "user": user(&self.author), "assignees": self.assignees.iter().map(|a| user(a)).collect::>(), "head": { "sha": self.head }, "base": { "ref": push.branch }, "draft": self.draft, "mergeable": action == "mergeable", }, "repository": { "id": push.repo_id, "full_name": push.repo }, }); serde_json::to_vec(&body).unwrap_or_default() } } /// Read the open pull requests on `push`'s branch with `list`, up to /// [`MAX_READS`] times [`RECHECK`] apart, and `publish` what each says: the /// action `mergeable` the first read that finds it so, and `conflicting` the /// second read in a row that finds the forge's flag false. A pull request that /// is a draft, or never settles, gets nothing. pub(crate) async fn check(push: &BranchPush, recheck: Duration, list: L, publish: P) where L: Fn() -> LF, LF: Future>>, P: Fn(&'static str, Vec) -> PF, PF: Future, { let mut after_false = HashSet::new(); let mut settled = HashSet::new(); for read in 0..MAX_READS { if read > 0 { tokio::time::sleep(recheck).await; } let pulls = match list().await { Ok(pulls) => pulls, Err(e) => { tracing::warn!(repo = %push.repo, branch = %push.branch, error = ?e, "pr conflicts: listing pull requests failed"); return; } }; for pull in &pulls { if pull.draft || settled.contains(&pull.number) { continue; } let action = match pull.mergeable { Some(true) => "mergeable", Some(false) if after_false.contains(&pull.number) => "conflicting", Some(false) => { after_false.insert(pull.number); continue; } None => { after_false.remove(&pull.number); continue; } }; settled.insert(pull.number); publish(action, pull.event_body(push, action)).await; } if pulls.iter().all(|p| p.draft || settled.contains(&p.number)) { return; } } } /// 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 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 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 { return; }; loop { tokio::time::sleep(SETTLE).await; if let Some(forge) = state.forge.get() { check( &push, RECHECK, || async { 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| { let (state, push) = (&state, &push); async move { if let Err(not_kept) = crate::forge_events::publish_event(state, MERGEABILITY_EVENT, body) .await { tracing::warn!(repo = %push.repo, action, reason = ?not_kept, "pr conflicts: event not kept"); } } }, ) .await; } if !running.again() { return; } } } #[cfg(test)] mod tests { use super::*; use std::sync::{Arc, Mutex}; fn push() -> BranchPush { BranchPush { repo: "o/r".to_owned(), repo_id: 7, branch: "main".to_owned(), default_branch: "main".to_owned(), } } fn pull(number: i64, mergeable: Option) -> Pull { Pull { number, title: "t".to_owned(), url: "https://forge/o/r/pulls/1".to_owned(), author: "atlas".to_owned(), assignees: BTreeSet::from(["iris".to_owned()]), head: "abc".to_owned(), draft: false, mergeable, } } /// What `check` published for the reads `reads` give in turn, as /// `(action, number)`, and how many reads it made. async fn run(reads: Vec>) -> (Vec<(String, i64)>, usize) { let seen = Arc::new(Mutex::new(Vec::new())); let calls = Arc::new(Mutex::new(0_usize)); let reads = Arc::new(reads); check( &push(), Duration::ZERO, || { let (reads, calls) = (Arc::clone(&reads), Arc::clone(&calls)); async move { let mut n = calls.lock().expect("lock"); let read = reads.get(*n).or(reads.last()).cloned().unwrap_or_default(); *n += 1; Ok(read) } }, |action, body| { let seen = Arc::clone(&seen); async move { let json: Json = serde_json::from_slice(&body).expect("json"); assert_eq!(json["action"], action); seen.lock() .expect("lock") .push((action.to_owned(), json["number"].as_i64().expect("number"))); } }, ) .await; let published = seen.lock().expect("lock").clone(); let made = *calls.lock().expect("lock"); (published, made) } fn act(a: &str, n: i64) -> (String, i64) { (a.to_owned(), n) } #[tokio::test] async fn each_pull_request_is_published_by_what_the_forge_says() { let (published, reads) = run(vec![vec![pull(1, Some(true)), pull(2, Some(false))]]).await; assert_eq!(published[0], act("mergeable", 1)); assert_eq!(reads, 2, "the conflicting one is read again to confirm it"); assert_eq!(published.len(), 2); assert_eq!(published[1], act("conflicting", 2)); } #[tokio::test] async fn all_mergeable_pull_requests_cost_one_read() { let (published, reads) = run(vec![vec![pull(1, Some(true)), pull(2, Some(true))]]).await; assert_eq!(published, [act("mergeable", 1), act("mergeable", 2)]); assert_eq!(reads, 1); let (none, reads) = run(vec![vec![]]).await; assert!(none.is_empty()); assert_eq!(reads, 1); } /// A check still running reads as false; it must not become a conflict. #[tokio::test] async fn a_check_still_running_is_not_a_conflict() { let (published, reads) = run(vec![vec![pull(1, Some(false))], vec![pull(1, Some(true))]]).await; assert_eq!(published, [act("mergeable", 1)]); assert_eq!(reads, 2); } #[tokio::test] async fn a_pull_request_the_forge_has_not_decided_is_read_again() { let (published, reads) = run(vec![ vec![pull(1, None)], vec![pull(1, None)], vec![pull(1, Some(true))], ]) .await; assert_eq!(published, [act("mergeable", 1)]); assert_eq!(reads, 3); let (published, reads) = run(vec![vec![pull(1, None)]]).await; assert!(published.is_empty(), "never decided, never published"); assert_eq!(reads, MAX_READS); // False, then undecided, then false is not two false reads in a row. let (published, _) = run(vec![ vec![pull(1, Some(false))], vec![pull(1, None)], vec![pull(1, Some(false))], vec![pull(1, None)], ]) .await; assert!(published.is_empty()); } #[tokio::test] async fn a_draft_is_left_alone() { let draft = Pull { draft: true, ..pull(1, Some(false)) }; let (published, reads) = run(vec![vec![draft.clone()]]).await; assert!(published.is_empty()); assert_eq!(reads, 1); // Beside a pull request that needs a second read, the draft would // read false twice, as it always does. let (published, reads) = run(vec![vec![draft, pull(2, Some(false))]]).await; assert_eq!(published, [act("conflicting", 2)]); assert_eq!(reads, 2); } #[tokio::test] async fn a_failed_listing_publishes_nothing() { let published = Arc::new(Mutex::new(0_usize)); let counted = Arc::clone(&published); check( &push(), Duration::ZERO, || async { anyhow::bail!("forge down") }, |_, _| { let counted = Arc::clone(&counted); async move { *counted.lock().expect("lock") += 1; } }, ) .await; assert_eq!(*published.lock().expect("lock"), 0); } /// The body is a `pull_request` delivery in the shape the router parses. #[test] fn the_event_body_names_the_author_the_assignees_and_the_base() { let body = pull(3, Some(false)).event_body(&push(), "conflicting"); let json: Json = serde_json::from_slice(&body).expect("json"); assert_eq!(json["action"], "conflicting"); assert_eq!(json["repository"]["id"], 7); assert_eq!(json["repository"]["full_name"], "o/r"); assert_eq!(json["pull_request"]["number"], 3); assert_eq!(json["pull_request"]["user"]["login"], "atlas"); assert_eq!(json["pull_request"]["assignees"][0]["login"], "iris"); assert_eq!(json["pull_request"]["base"]["ref"], "main"); assert_eq!(json["pull_request"]["head"]["sha"], "abc"); } #[test] fn a_listed_pull_request_is_read_and_one_without_an_author_is_not() { let listed = |user: Json| -> PullRequest { serde_json::from_value(json!({ "number": 4, "title": "t", "html_url": "https://forge/o/r/pulls/4", "user": user, "assignees": [{ "login": "iris" }], "head": { "sha": "abc" }, "draft": false, "mergeable": false, "labels": [], "requested_reviewers": [], "base": { "ref": "main" }, "created_at": "2026-11-07T08:00:00Z", "updated_at": "2026-11-07T09:00:00Z", "additions": 1, "deletions": 1, })) .expect("a pull request") }; let got = Pull::from_listed(&listed(json!({ "login": "atlas" }))).expect("read"); assert_eq!(got.number, 4); assert_eq!(got.author, "atlas"); assert_eq!(got.mergeable, Some(false)); assert_eq!(got.assignees, BTreeSet::from(["iris".to_owned()])); assert_eq!(Pull::from_listed(&listed(Json::Null)), None); } fn push_body(git_ref: &str, after: &str) -> Vec { serde_json::to_vec(&json!({ "ref": git_ref, "after": after, "repository": { "id": 7, "full_name": "o/r", "default_branch": "main" }, })) .expect("json") } #[test] fn a_push_to_a_branch_is_read_and_a_tag_or_a_delete_is_not() { assert_eq!( BranchPush::parse(&push_body("refs/heads/feat/x", "abc123")), Some(BranchPush { repo: "o/r".to_owned(), repo_id: 7, branch: "feat/x".to_owned(), default_branch: "main".to_owned(), }) ); assert_eq!( BranchPush::parse(&push_body("refs/tags/v1", "abc123")), None ); assert_eq!( BranchPush::parse(&push_body("refs/heads/main", "0000000")), None ); assert_eq!(BranchPush::parse(b"not json"), None); } /// Pushes during a check make it run once more, however many they are. #[test] fn pushes_during_a_check_are_one_more_check() { let d = Debounce::default(); let key = |b: &str| ("o/r".to_owned(), b.to_owned()); assert!(d.begin(key("main")), "the first push starts a check"); assert!(!d.begin(key("main"))); assert!(!d.begin(key("main"))); assert!(d.begin(key("other")), "another branch has its own"); assert!(d.finish(&key("main")), "the pushes since ask for one more"); assert!(!d.finish(&key("main")), "and then the branch is free"); 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"); } }