From 11eb7dac9c8228fd054f5283617613a633a725c8 Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 11 Nov 2026 04:36:28 +0100 Subject: [PATCH] an agent gets one todo when its pr starts conflicting with the target branch --- docs/4-implementation/swarm/controller.md | 4 + docs/4-implementation/swarm/notify-router.md | 55 +- swarm-controller/src/forge.rs | 5 +- swarm-controller/src/forge/base_pulls.rs | 42 ++ swarm-controller/src/forge_events.rs | 38 +- swarm-controller/src/main.rs | 1 + swarm-controller/src/pr_conflicts.rs | 517 +++++++++++++++++++ swarm-controller/src/webhook.rs | 4 + swarm-notify-router/src/event.rs | 99 ++++ swarm-notify-router/src/route.rs | 107 +++- swarm-notify-router/src/stored.rs | 8 +- swarm-wire/src/forge_events.rs | 6 + swarm-wire/src/notify_flows.rs | 98 +++- 13 files changed, 963 insertions(+), 21 deletions(-) create mode 100644 swarm-controller/src/forge/base_pulls.rs create mode 100644 swarm-controller/src/pr_conflicts.rs diff --git a/docs/4-implementation/swarm/controller.md b/docs/4-implementation/swarm/controller.md index b34919142..654918c39 100644 --- a/docs/4-implementation/swarm/controller.md +++ b/docs/4-implementation/swarm/controller.md @@ -384,6 +384,10 @@ itself, all under `https:///webhook/forge/`: whose deliveries it keeps in the `SWARM_EVENTS` stream for the [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 +to the same stream: [merge conflicts as a source](notify-router.md#flows). + The controller **interprets** a delivery and sends hives a specific message — _the knowledge repo changed_, _deploy agent `foo` at rev `abc123`_ — rather than forwarding forge payloads for each hive to diff --git a/docs/4-implementation/swarm/notify-router.md b/docs/4-implementation/swarm/notify-router.md index ba4af44d9..f6c5801cd 100644 --- a/docs/4-implementation/swarm/notify-router.md +++ b/docs/4-implementation/swarm/notify-router.md @@ -12,14 +12,16 @@ the way it did. An agent can get the same news from both. ## What an agent gets -| Flow | Event | The agent's todo | -| -------------- | ------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `assigned` | someone assigns an agent to an issue or PR | `you are assigned to issue #: (<url>)`, key `forge:<repo>#<n>:assignment` | -| `unassigned` | someone takes an agent off an issue or PR | the same key rewritten: `you are not assigned to issue <repo>#<n> anymore: <title> (<url>)` | -| `closed` | someone closes or merges the issue or PR | the router clears every router todo it queued for it, then queues `closed` on each agent that held one: `<kind> <repo>#<n> was closed: <title> (<url>)`, key `forge:<repo>#<n>:closed` | -| `ci-failed` | a CI run on a PR finishes with a failed job | the PR's author gets `CI failed on PR <repo>#<n>: <title> (<url>)`, key `forge:<repo>#<n>:ci` | -| `ci-cancelled` | a CI run on a PR gets cancelled | the same key rewritten: `CI was cancelled on PR <repo>#<n>: <title> (<url>)` | -| `ci-cleared` | someone pushes to the PR's branch | the router clears the PR author's `:ci` todo, as the push starts a new run; the author's other todos for the PR stay | +| Flow | Event | The agent's todo | +| ------------------ | ---------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `assigned` | someone assigns an agent to an issue or PR | `you are assigned to issue <repo>#<n>: <title> (<url>)`, key `forge:<repo>#<n>:assignment` | +| `unassigned` | someone takes an agent off an issue or PR | the same key rewritten: `you are not assigned to issue <repo>#<n> anymore: <title> (<url>)` | +| `closed` | someone closes or merges the issue or PR | the router clears every router todo it queued for it, then queues `closed` on each agent that held one: `<kind> <repo>#<n> was closed: <title> (<url>)`, key `forge:<repo>#<n>:closed` | +| `ci-failed` | a CI run on a PR finishes with a failed job | the PR's author gets `CI failed on PR <repo>#<n>: <title> (<url>)`, key `forge:<repo>#<n>:ci` | +| `ci-cancelled` | a CI run on a PR gets cancelled | the same key rewritten: `CI was cancelled on PR <repo>#<n>: <title> (<url>)` | +| `ci-cleared` | someone pushes to the PR's branch | the router clears the PR author's `:ci` todo, as the push starts a new run; the author's other todos for the PR stay | +| `conflict` | the controller finds that a PR no longer merges into its target branch | the PR's author gets `PR <repo>#<n> no longer merges into its target branch: <title> (<url>)`, key `forge:<repo>#<n>:conflict`, once until the PR merges again | +| `conflict-cleared` | the controller finds that a conflicting PR merges again | the router clears the PR author's `:conflict` todo | The todos are subsystem `router` in the agent's todo list. A todo under the same key replaces the one before, so an issue gives an agent one assignment @@ -85,8 +87,8 @@ under a `todo` or a `request_reviewer`. The most is 64 nodes, 16 levels deep, and 100 custom flows. A `todo` node sends to the agents an event field holds (`assignees_added`, `todo_holders`, …) or to one named agent. -The router ships the six flows above, each a source → match on `action` → -todo, and `head-classifier`: source `pull_request` and `pull_request_sync` → +The router ships the eight flows above, each a source → match on `action` → +todo (`conflict` and `conflict-cleared` also match `mergeability_changed`), and `head-classifier`: source `pull_request` and `pull_request_sync` → match on `action` (`opened`, `reopened`, `synchronized`) and `draft` = `false`, or on `undrafted` → `request_reviewer` for `review-classifier` ([below](#requesting-a-review)). The code defines them (`swarm_wire::notify_flows::builtins`), so a build that changes one reaches @@ -100,7 +102,7 @@ switched off, and the custom flows. Only the controller writes it, through compare-and-set. The controller answers 400 unless every flow passes the tree check above and reads only what exists: a source event from `issues`, `issue_assign`, `pull_request`, `pull_request_assign`, `pull_request_sync` or one of the -three `pull_request_review_*` events, `workflow_run_failure` or `workflow_run_cancelled`, and condition and recipient fields from the event's field list. The same goes for the `{{ field }}` +three `pull_request_review_*` events, `workflow_run_failure`, `workflow_run_cancelled` or `pull_request_mergeability`, and condition and recipient fields from the event's field list. The same goes for the `{{ field }}` variables in a todo's key and text, which are Jinja templates (see [`hive-template`](../../../hive-template/README.md)). A `PUT` must carry the `revision` it read, `0` for no document, and gets 409 when the stored revision is another. Swarm UI's [notification router card on `/swarm`](../web-ui/swarm-ui.md#notification-flows) shows the document and @@ -213,6 +215,35 @@ head commit from the last delivery about the PR that wasn't a run. A run whose delivery names another head began before the latest push, and the router drops 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 +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. + +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 +merge it's still one; each PR that reads false or undecided adds a second read +at least, and four is the most. + +The event also carries `mergeability_changed`, which the router sets to `true` +only when the event moves the PR into or out of conflict, judged against the +`conflicting` flag it keeps per PR in its snapshot. `conflict` and +`conflict-cleared` match on it, so a PR that stays in conflict across many +pushes to its target branch gets the one todo, and a PR that merges and never +conflicted gets nothing. The flag resets when the PR closes, so the router tells a reopened PR +that still conflicts again. A PR whose author fixed it stays flagged +until the next push to its target branch finds it merging. + **What the router runs.** It reads the key at most every 10 seconds, when an event arrives, and runs the built-ins you left on, then the custom flows that are on and pass the tree check. It logs a custom flow that fails the check, @@ -239,7 +270,7 @@ holding such a node (a custom flow with one) as one it can't read and keeps the flows it had; it also has no `head-classifier`. A router that has `head-classifier` but predates the `draft` and `undrafted` fields requests the review of a draft PR as before. A router that has `request_reviewer`, with a -controller that predates it, gets no answer to the request and logs that. A controller that predates this leaves the key absent, and the +controller that predates it, gets no answer to the request and logs that. A router that predates the `conflict` flows ignores the `pull_request_mergeability` events a newer controller sends, and gives no conflict todos; a controller that predates them sends none, so the new flows never run. A controller that predates this leaves the key absent, and the router runs the built-ins as before. A router that predates this ignores the document, so the page and the API show a state the router isn't yet running. diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 521e1a859..854bcdcc0 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -46,6 +46,7 @@ use swarm_wire::review_classifier::Commit; pub mod agent_token; mod api; +mod base_pulls; pub mod builder_account; pub mod ci_metrics_account; pub mod classifier_account; @@ -2321,10 +2322,12 @@ mod tests { ["workflow_run_failure", "workflow_run_cancelled"] ); // The events a stored flow's source may name are the ones the hooks - // deliver to the router. + // deliver to the router, and the one the controller makes. let mut routed = events(DeliveryKind::Issues); routed.extend(events(DeliveryKind::PullRequest)); routed.extend(events(DeliveryKind::WorkflowRun)); + // Not a hook: the controller publishes it itself (`crate::pr_conflicts`). + routed.push(swarm_wire::forge_events::MERGEABILITY_EVENT); routed.sort_unstable(); let mut sources = swarm_wire::notify_flows::SOURCE_EVENTS .iter() diff --git a/swarm-controller/src/forge/base_pulls.rs b/swarm-controller/src/forge/base_pulls.rs new file mode 100644 index 000000000..4a59bcef2 --- /dev/null +++ b/swarm-controller/src/forge/base_pulls.rs @@ -0,0 +1,42 @@ +//! The open pull requests that target one branch. + +use anyhow::{Context, Result}; +use forgejo_api::structs::{ + PullRequest, RepoListPullRequestsQuery, RepoListPullRequestsQueryState, +}; + +use super::Client; + +impl Client { + /// Every open pull request in `owner/repo` whose target branch is `base`, + /// as the forge lists them: one call per page of 50. + /// + /// # Errors + /// + /// The forge refused or did not answer. + pub async fn open_pulls_on( + &self, + owner: &str, + repo: &str, + base: &str, + ) -> Result<Vec<PullRequest>> { + let query = RepoListPullRequestsQuery { + state: Some(RepoListPullRequestsQueryState::Open), + sort: None, + milestone: None, + labels: None, + poster: None, + base: Some(base.to_owned()), + head: None, + }; + self.api + .call(|api| { + Box::pin( + api.repo_list_pull_requests(owner, repo, query.clone()) + .all(), + ) + }) + .await + .with_context(|| format!("list open pull requests on {base} in {owner}/{repo}")) + } +} diff --git a/swarm-controller/src/forge_events.rs b/swarm-controller/src/forge_events.rs index dd0b4ea9d..d1000f8ad 100644 --- a/swarm-controller/src/forge_events.rs +++ b/swarm-controller/src/forge_events.rs @@ -10,6 +10,9 @@ //! `pull_request_sync`, …), not `X-Forgejo-Event`, which folds those into //! `issues` and `pull_request`. It reaches the subject only once it is one //! plain token. +//! +//! [`publish_event`] keeps an event this daemon makes, such as the +//! `pull_request_mergeability` event of `crate::pr_conflicts`, the same way. use axum::http::HeaderMap; use std::time::Duration; @@ -51,15 +54,42 @@ pub(crate) async fn publish( body: axum::body::Bytes, ) -> Result<(), NotKept> { let subject = subject(headers)?; + let mut message = async_nats::jetstream::message::PublishMessage::build().payload(body); + if let Some(id) = headers.get(DELIVERY_HEADER).and_then(|v| v.to_str().ok()) { + message = message.message_id(id); + } + keep(state, subject, message).await +} + +/// Keep an event this daemon makes itself, of `event_type`, in the stream, and +/// wait for the stream's ack. The router reads it like a delivery. +pub(crate) async fn publish_event( + state: &crate::routes::AppState, + event_type: &str, + body: Vec<u8>, +) -> Result<(), NotKept> { + if !swarm_wire::forge_events::is_subject_token(event_type) { + return Err(NotKept::BadEventType); + } + let message = async_nats::jetstream::message::PublishMessage::build().payload(body.into()); + keep( + state, + swarm_wire::forge_events::subject(event_type), + message, + ) + .await +} + +async fn keep( + state: &crate::routes::AppState, + subject: String, + message: async_nats::jetstream::message::PublishMessage, +) -> Result<(), NotKept> { let Some(status) = state.status.as_ref() else { return Err(NotKept::Queue("no swarm queue is configured".to_owned())); }; let client = status.queue_client(); swarm_queue_client::ensure_connected(&client).map_err(|e| NotKept::Queue(e.to_string()))?; - let mut message = async_nats::jetstream::message::PublishMessage::build().payload(body); - if let Some(id) = headers.get(DELIVERY_HEADER).and_then(|v| v.to_str().ok()) { - message = message.message_id(id); - } let kept = async { swarm_queue_client::forge_events::open_or_create(&client) .await diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 0da2f2bd6..cdcd2fbfd 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -72,6 +72,7 @@ mod notify_router_api; mod observability_grant; mod otel_http_client; mod pause_hold; +mod pr_conflicts; mod pr_report; mod proxy_credential; mod pulls; diff --git a/swarm-controller/src/pr_conflicts.rs b/swarm-controller/src/pr_conflicts.rs new file mode 100644 index 000000000..1d3769153 --- /dev/null +++ b/swarm-controller/src/pr_conflicts.rs @@ -0,0 +1,517 @@ +//! 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 a branch 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, +} + +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<Self> { + 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(), + }) + } + + 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<HashMap<(String, String), bool>>, +} + +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 + } + + /// 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<Debounce> = 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<String>, + head: String, + draft: bool, + /// The forge's flag: `None` when it did not say. + mergeable: Option<bool>, +} + +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<Self> { + 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<u8> { + 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::<Vec<_>>(), + "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<L, LF, P, PF>(push: &BranchPush, recheck: Duration, list: L, publish: P) +where + L: Fn() -> LF, + LF: Future<Output = anyhow::Result<Vec<Pull>>>, + P: Fn(&'static str, Vec<u8>) -> PF, + PF: Future<Output = ()>, +{ + 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; + } + } +} + +/// 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. +pub(crate) fn on_push(state: &AppState, body: &[u8]) { + let Some(push) = BranchPush::parse(body) else { + return; + }; + if RUNNING.begin(push.key()) { + tokio::spawn(watch(state.clone(), push)); + } +} + +async fn watch(state: AppState, push: BranchPush) { + let Some((owner, name)) = push.repo.split_once('/') else { + RUNNING.finish(&push.key()); + return; + }; + loop { + tokio::time::sleep(SETTLE).await; + if let Some(forge) = state.forge.get() { + check( + &push, + RECHECK, + || async { + let listed = 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.finish(&push.key()) { + 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(), + } + } + + fn pull(number: i64, mergeable: Option<bool>) -> 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<Pull>>) -> (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<u8> { + serde_json::to_vec(&json!({ + "ref": git_ref, + "after": after, + "repository": { "id": 7, "full_name": "o/r" }, + })) + .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(), + }) + ); + 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"))); + } +} diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index e0a016f94..4aee907a5 100644 --- a/swarm-controller/src/webhook.rs +++ b/swarm-controller/src/webhook.rs @@ -406,6 +406,9 @@ pub(super) enum DeliveryKind { /// hook: that one is repo-scoped to the knowledge repo alone, and this /// 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 + /// (`crate::pr_conflicts`). #[strum(serialize = "vcs-activity")] VcsActivity, /// `pull_request_review_request` events on every repo in the instance, @@ -647,6 +650,7 @@ pub(super) async fn post_webhook_forge( tokio::spawn(crate::vcs_metrics::count_default_branch_push( activity, forge, )); + crate::pr_conflicts::on_push(&state, &body); } else { tracing::warn!("webhook: vcs-activity delivery did not parse as a push payload"); } diff --git a/swarm-notify-router/src/event.rs b/swarm-notify-router/src/event.rs index 1bb798337..06664d534 100644 --- a/swarm-notify-router/src/event.rs +++ b/swarm-notify-router/src/event.rs @@ -13,6 +13,8 @@ use std::collections::BTreeSet; use serde::{Deserialize, Serialize}; use serde_json::Value as Json; +use swarm_wire::forge_events::MERGEABILITY_EVENT; + use crate::flow::{Fields, Value}; /// The parts of an `issues` or `pull_request` family delivery the router @@ -165,6 +167,41 @@ pub struct Snapshot { /// and tells nobody anything. #[serde(default)] pub head: Option<String>, + /// Whether the last mergeability event for this PR found it conflicting + /// with its target branch, or `None` before the first. + #[serde(default)] + pub conflicting: Option<bool>, +} + +/// Whether a mergeability event moves the PR into or out of conflict: a +/// `conflicting` finding for a PR not already conflicting, or a `mergeable` +/// one for a PR that was. +fn mergeability_changed(event_type: &str, action: &str, snapshot: &Snapshot) -> bool { + if event_type != MERGEABILITY_EVENT { + return false; + } + match action { + "conflicting" => snapshot.conflicting != Some(true), + "mergeable" => snapshot.conflicting == Some(true), + _ => false, + } +} + +/// The PR's conflict state after `event_type`, or `snapshot`'s when the event +/// says nothing about it. A closed PR forgets it: the `closed` flow clears the +/// conflict todo, so a reopened PR that still conflicts is told again. +pub fn conflicting_after(event_type: &str, action: &str, snapshot: &Snapshot) -> Option<bool> { + if event_type == "pull_request" && action == "closed" { + return None; + } + if event_type != MERGEABILITY_EVENT { + return snapshot.conflicting; + } + match action { + "conflicting" => Some(true), + "mergeable" => Some(false), + _ => snapshot.conflicting, + } } /// Whether `event_type` is an assign delivery, whose assignee list is the @@ -199,6 +236,14 @@ pub fn fields(event_type: &str, delivery: &Delivery, snapshot: &Snapshot) -> Fie ("author", text(&delivery.author)), ("draft", text(bool_text(delivery.draft))), ("undrafted", text(bool_text(delivery.undrafted))), + ( + "mergeability_changed", + text(bool_text(mergeability_changed( + event_type, + &delivery.action, + snapshot, + ))), + ), ( "assignees", Value::List(delivery.assignees.iter().cloned().collect()), @@ -428,6 +473,60 @@ mod tests { assert_eq!(f.get("undrafted"), Some(&Value::Text("true".to_owned()))); } + /// The body swarm-controller publishes for a mergeability finding reads as + /// a PR, and `mergeability_changed` is true only for a change. + #[test] + fn a_mergeability_event_reads_as_a_pull_request_and_flags_only_a_change() { + let body = |action: &str| { + serde_json::to_vec(&serde_json::json!({ + "action": action, + "number": 9, + "pull_request": { + "number": 9, + "title": "t", + "html_url": "https://forge/pulls/9", + "user": { "login": "atlas" }, + "assignees": [{ "login": "iris" }], + "head": { "sha": "abc" }, + "base": { "ref": "main" }, + "draft": false, + }, + "repository": { "id": 42, "full_name": "o/r" }, + })) + .expect("encodes") + }; + let d = Delivery::from_event(MERGEABILITY_EVENT, &body("conflicting")).expect("parses"); + assert!(d.is_pull); + assert_eq!( + (d.author.as_str(), d.number, d.action.as_str()), + ("atlas", 9, "conflicting") + ); + assert_eq!(d.assignees, BTreeSet::from(["iris".to_owned()])); + + let changed = |action: &str, before: Option<bool>, event: &str| { + let d = Delivery::from_event(event, &body(action)).expect("parses"); + let snapshot = Snapshot { + conflicting: before, + ..Snapshot::default() + }; + fields(event, &d, &snapshot) + .get("mergeability_changed") + .cloned() + }; + let yes = Some(Value::Text("true".to_owned())); + let no = Some(Value::Text("false".to_owned())); + assert_eq!(changed("conflicting", None, MERGEABILITY_EVENT), yes); + assert_eq!(changed("conflicting", Some(false), MERGEABILITY_EVENT), yes); + assert_eq!(changed("conflicting", Some(true), MERGEABILITY_EVENT), no); + assert_eq!(changed("mergeable", Some(true), MERGEABILITY_EVENT), yes); + assert_eq!(changed("mergeable", Some(false), MERGEABILITY_EVENT), no); + assert_eq!(changed("mergeable", None, MERGEABILITY_EVENT), no); + assert_eq!( + changed("conflicting", None, "pull_request"), + no, + "another event type never changes it" + ); + } #[test] fn the_snapshot_turns_the_after_list_into_added_and_removed() { let d = Delivery::parse(&issue_assign("unassigned", &["iris"])).expect("parses"); diff --git a/swarm-notify-router/src/route.rs b/swarm-notify-router/src/route.rs index c5765357c..7be7a321d 100644 --- a/swarm-notify-router/src/route.rs +++ b/swarm-notify-router/src/route.rs @@ -5,7 +5,7 @@ use std::collections::BTreeSet; use swarm_wire::router_todo::Op; -use crate::event::{Delivery, Snapshot, fields, is_assign}; +use crate::event::{Delivery, Snapshot, conflicting_after, fields, is_assign}; use swarm_wire::notify_flows::Flow; use crate::flow::{Action, ReviewRequest, evaluate, requests}; @@ -67,6 +67,7 @@ pub fn route( if !is_run && delivery.is_pull && !delivery.head.is_empty() { next.head = Some(delivery.head.clone()); } + next.conflicting = conflicting_after(event_type, &delivery.action, snapshot); if is_assign(event_type) { // A skipped assignee is left out of the snapshot, so the next assign // event counts it as added again: an agent assigned before it is in @@ -204,6 +205,110 @@ mod tests { d } + /// A PR starts conflicting: its author gets one todo, a second finding of + /// the same changes nothing, and the PR merging again clears it. Closing + /// the PR clears it too and forgets the state, so a reopened PR that + /// still conflicts is told again. + #[test] + fn a_pull_request_that_starts_conflicting_tells_the_author_once() { + let flows = builtins(); + let is_agent = |a: &str| a == "atlas"; + let found = |action: &str, snapshot: &Snapshot| { + route( + "pull_request_mergeability", + &pull_by("atlas", action), + snapshot, + &flows, + is_agent, + ) + }; + + let first = found("conflicting", &Snapshot::default()); + assert_eq!(agents(&first), [("atlas", Op::Upsert)]); + assert_eq!(first.actions[0].op.key, "forge:o/r#9:conflict"); + assert_eq!(first.actions[0].flow, "conflict"); + assert!( + first.actions[0] + .op + .summary + .starts_with("PR o/r#9 no longer merges into its target branch") + ); + assert_eq!(first.next.conflicting, Some(true)); + assert_eq!(first.next.holders, BTreeSet::from(["atlas".to_owned()])); + + let again = found("conflicting", &first.next); + assert!(again.actions.is_empty(), "no todo for a repeat: {again:?}"); + assert_eq!(again.next.conflicting, Some(true)); + + let merges = found("mergeable", &again.next); + assert_eq!(agents(&merges), [("atlas", Op::Clear)]); + assert_eq!(merges.actions[0].op.key, "forge:o/r#9:conflict"); + assert!(!merges.actions[0].op.prefix); + assert_eq!(merges.next.conflicting, Some(false)); + + let later = found("conflicting", &merges.next); + assert_eq!( + agents(&later), + [("atlas", Op::Upsert)], + "a PR that conflicts again after merging is told again" + ); + + let closed = route( + "pull_request", + &pull_by("atlas", "closed"), + &later.next, + &flows, + is_agent, + ); + assert!( + closed + .actions + .iter() + .any(|a| a.op.op == Op::Clear && a.op.prefix && a.op.key == "forge:o/r#9:"), + "the close clears every todo for the PR, the conflict one included" + ); + assert_eq!(closed.next.conflicting, None); + assert_eq!( + agents(&found("conflicting", &closed.next)), + [("atlas", Op::Upsert)] + ); + } + + /// A PR that merges and never conflicted has nothing to clear, and the + /// mergeability events leave the other state alone. + #[test] + fn a_mergeable_finding_on_a_pull_request_that_never_conflicted_does_nothing() { + let flows = builtins(); + for before in [None, Some(false)] { + let snapshot = Snapshot { + conflicting: before, + ..Snapshot::default() + }; + let r = route( + "pull_request_mergeability", + &pull_by("atlas", "mergeable"), + &snapshot, + &flows, + |_| true, + ); + assert!(r.actions.is_empty(), "{before:?}: {:?}", r.actions); + assert_eq!(r.next.conflicting, Some(false)); + } + // Control: the same finding on a conflicting PR does clear. + let conflicting = Snapshot { + conflicting: Some(true), + ..Snapshot::default() + }; + let r = route( + "pull_request_mergeability", + &pull_by("atlas", "mergeable"), + &conflicting, + &flows, + |_| true, + ); + assert_eq!(agents(&r), [("atlas", Op::Clear)]); + } + /// The push that clears the CI todo also cancels the runs before it, and /// one of them can report in after the clear. It began at the old head, so /// it must not bring the todo back. diff --git a/swarm-notify-router/src/stored.rs b/swarm-notify-router/src/stored.rs index 904354a43..7aa640961 100644 --- a/swarm-notify-router/src/stored.rs +++ b/swarm-notify-router/src/stored.rs @@ -150,7 +150,7 @@ mod tests { use super::*; /// The built-in flows, in the order the router runs them. - const BUILTIN_IDS: [&str; 7] = [ + const BUILTIN_IDS: [&str; 9] = [ "assigned", "unassigned", "closed", @@ -158,6 +158,8 @@ mod tests { "ci-failed", "ci-cancelled", "ci-cleared", + "conflict", + "conflict-cleared", ]; fn ids(f: &Flows) -> Vec<&str> { @@ -220,6 +222,8 @@ mod tests { "ci-failed", "ci-cancelled", "ci-cleared", + "conflict", + "conflict-cleared", "triage" ] ); @@ -232,7 +236,7 @@ mod tests { let before = ids(&f).join(","); assert_eq!( before, - "assigned,unassigned,head-classifier,ci-failed,ci-cancelled,ci-cleared,triage" + "assigned,unassigned,head-classifier,ci-failed,ci-cancelled,ci-cleared,conflict,conflict-cleared,triage" ); for bad in [ junk(2, b"not json"), diff --git a/swarm-wire/src/forge_events.rs b/swarm-wire/src/forge_events.rs index c0ad48506..f99b4aadf 100644 --- a/swarm-wire/src/forge_events.rs +++ b/swarm-wire/src/forge_events.rs @@ -24,6 +24,12 @@ pub const MAX_AGE: Duration = Duration::from_hours(7 * 24); /// The router's durable consumer on [`STREAM`]. pub const ROUTER_CONSUMER: &str = "notify-router"; +/// The event type swarm-controller publishes, on [`subject`], for a pull request +/// it has checked against its target branch after a push there. Forgejo sends +/// no such delivery. The body is a `pull_request` delivery whose `action` is +/// `conflicting` or `mergeable`. +pub const MERGEABILITY_EVENT: &str = "pull_request_mergeability"; + /// The subject a delivery of Forgejo event type `event_type` is kept under. #[must_use] pub fn subject(event_type: &str) -> String { diff --git a/swarm-wire/src/notify_flows.rs b/swarm-wire/src/notify_flows.rs index b64840fba..de80c3929 100644 --- a/swarm-wire/src/notify_flows.rs +++ b/swarm-wire/src/notify_flows.rs @@ -88,6 +88,11 @@ pub const SOURCE_EVENTS: &[(&str, &str)] = &[ "workflow_run_cancelled", "A CI run on a pull request that was cancelled", ), + ( + "pull_request_mergeability", + "What swarm-controller found about an open pull request after a push to its target \ + branch: action conflicting, or mergeable", + ), ]; /// The fields of an event a condition or a recipient can name, with what each @@ -123,6 +128,10 @@ pub const FIELDS: &[(&str, &str)] = &[ "todo_holders", "The agents holding a router todo for the issue or pull request (a list)", ), + ( + "mergeability_changed", + "true when a mergeability event moves the pull request into or out of conflict, else false", + ), ]; /// The forge account of the review classifier, whose review the built-in @@ -426,6 +435,11 @@ const ACTIONS: &[(&str, &str)] = &[ ("assigned", "An assignee added"), ("unassigned", "An assignee removed"), ("synchronized", "New commits pushed to a pull request"), + ( + "conflicting", + "A pull request does not merge into its target branch", + ), + ("mergeable", "A pull request merges into its target branch"), ]; /// One event field a condition or recipient can name, and what it can hold. @@ -838,6 +852,10 @@ impl TodoSink { /// later run's outcome replaces the earlier one. /// - `ci-cleared`: a push to the PR's branch clears that todo, as the new run /// replaces the old one. +/// - `conflict` and `conflict-cleared`: a PR that stops merging into its +/// target branch gives its author a todo, and the todo clears when it merges +/// again. Both read `mergeability_changed`, so a repeated finding that the +/// PR still conflicts writes nothing. #[must_use] pub fn builtins() -> Vec<Flow> { let assignment_key = "forge:{{ repo }}#{{ number }}:assignment"; @@ -871,9 +889,47 @@ pub fn builtins() -> Vec<Flow> { head_classifier(), ]; flows.extend(ci_flows()); + flows.extend(conflict_flows()); flows } +/// source(`pull_request_mergeability`) → match(`action` and +/// `mergeability_changed` = `true`) → the author's `:conflict` todo. +fn conflict_flows() -> Vec<Flow> { + let flow = |id: &str, action: &str, op, text: &str| { + let mut flow = chain( + id, + &["pull_request_mergeability"], + action, + TodoSink { + to: Recipients::Field("author".to_owned()), + op, + key: "forge:{{ repo }}#{{ number }}:conflict".to_owned(), + prefix: false, + text: text.to_owned(), + }, + ); + for node in &mut flow.nodes { + if let NodeKind::Match { when } = &mut node.kind { + when.push(Condition { + field: "mergeability_changed".to_owned(), + equals: "true".to_owned(), + }); + } + } + flow + }; + vec![ + flow( + "conflict", + "conflicting", + Op::Upsert, + "{{ kind }} {{ repo }}#{{ number }} no longer merges into its target branch: {{ title }} ({{ url }})", + ), + flow("conflict-cleared", "mergeable", Op::Clear, ""), + ] +} + /// The flows that tell a PR's author about its CI: the todo for a failed run, /// the same key rewritten for a cancelled one, and the clear a push makes. fn ci_flows() -> Vec<Flow> { @@ -1077,6 +1133,42 @@ mod tests { } } + /// Both conflict flows act only when the mergeability event changed the + /// PR's state, and write the author's `:conflict` key. + #[test] + fn the_conflict_flows_read_the_change_and_write_the_conflict_key() { + for (id, action, op) in [ + ("conflict", "conflicting", Op::Upsert), + ("conflict-cleared", "mergeable", Op::Clear), + ] { + let f = builtin(id); + assert_eq!( + f.node("source").map(|n| &n.kind), + Some(&NodeKind::Source { + events: vec!["pull_request_mergeability".to_owned()] + }), + "{id}" + ); + let want = |field: &str, equals: &str| Condition { + field: field.to_owned(), + equals: equals.to_owned(), + }; + assert_eq!( + f.node("match").map(|n| &n.kind), + Some(&NodeKind::Match { + when: vec![want("action", action), want("mergeability_changed", "true")] + }), + "{id}" + ); + let Some(NodeKind::Todo(sink)) = f.node("sink").map(|n| &n.kind) else { + panic!("{id}: no todo"); + }; + assert_eq!(sink.op, op, "{id}"); + assert_eq!(sink.key, "forge:{{ repo }}#{{ number }}:conflict", "{id}"); + assert_eq!(sink.to, Recipients::Field("author".to_owned()), "{id}"); + } + } + /// What the schema offers is what the check accepts, and nothing else is. #[test] fn the_schema_lists_what_a_flow_may_name() { @@ -1519,7 +1611,9 @@ mod tests { "head-classifier", "ci-failed", "ci-cancelled", - "ci-cleared" + "ci-cleared", + "conflict", + "conflict-cleared" ], "an empty document is the built-ins" ); @@ -1541,6 +1635,8 @@ mod tests { "ci-failed", "ci-cancelled", "ci-cleared", + "conflict", + "conflict-cleared", "triage" ] );