diff --git a/docs/2-handbook/swarm/README.md b/docs/2-handbook/swarm/README.md index 1bd28715..6ad7f80d 100644 --- a/docs/2-handbook/swarm/README.md +++ b/docs/2-handbook/swarm/README.md @@ -437,8 +437,9 @@ itself, all under `https:///webhook/forge/`: the [review classifier](../../4-implementation/swarm/review-classifier.md#trigger-and-lifecycle) - instance-wide `issues` + `issue_assign` and `pull_request` + `pull_request_assign` + `pull_request_sync` + `pull_request_review_approved` + - `pull_request_review_rejected` + `pull_request_review_comment` system hooks, whose deliveries it - keeps in the `SWARM_EVENTS` stream for the + `pull_request_review_rejected` + `pull_request_review_comment` system hooks, and + an instance-wide `workflow_run_failure` + `workflow_run_cancelled` system hook, + whose deliveries it keeps in the `SWARM_EVENTS` stream for the [notification router](../../4-implementation/swarm/notify-router.md#the-path) The controller **interprets** a delivery and sends hives a specific diff --git a/docs/2-handbook/swarm/services.md b/docs/2-handbook/swarm/services.md index 379a0e67..bde977bb 100644 --- a/docs/2-handbook/swarm/services.md +++ b/docs/2-handbook/swarm/services.md @@ -595,11 +595,13 @@ that sets `deploy.notify-router.enable`. `allSwarmServices` defaults it on wherever the host has the router's store identity, on the review classifier's terms. Run one: it handles each issue's events in order. -swarm-controller keeps every `issues` and `pull_request` delivery from its -instance-wide hooks in the `SWARM_EVENTS` stream. The router reads that +swarm-controller keeps every `issues` and `pull_request` delivery, and every +failed or cancelled CI run, from its instance-wide hooks in the `SWARM_EVENTS` +stream. The router reads that stream and queues todos for the agents an event concerns: an agent assigned to an issue or pull request gets a todo, one taken off it gets that todo rewritten -to say so, and closing it clears them. A push to an open pull request makes +to say so, and closing it clears them. The author of a pull request gets a todo +when a CI run on it fails or gets cancelled, and a push to the branch clears it. A push to an open pull request makes it ask swarm-controller to request the review classifier's review. Each agent's harness takes its todos from the `ROUTER_TODOS` stream, so an agent that's down gets them when it's back. diff --git a/docs/4-implementation/swarm/notify-router.md b/docs/4-implementation/swarm/notify-router.md index 779c0dd2..2c994557 100644 --- a/docs/4-implementation/swarm/notify-router.md +++ b/docs/4-implementation/swarm/notify-router.md @@ -12,11 +12,14 @@ 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` | +| 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 | 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 @@ -39,12 +42,16 @@ router notices an unassign only on an issue it has seen an assign event for. ## The path -1. swarm-controller registers two instance-wide hooks, one for the `issues` - and `issue_assign` events at `/webhook/forge/issues` and one for +1. swarm-controller registers three instance-wide hooks, one for the `issues` + and `issue_assign` events at `/webhook/forge/issues`, one for `pull_request`, `pull_request_assign`, `pull_request_sync` (a push to an open PR's branch) and the submitted-review events `pull_request_review_approved`, - `pull_request_review_rejected` and `pull_request_review_comment` at `/webhook/forge/pull-request`. At each start it adds - any of these events a registered hook lacks. + `pull_request_review_rejected` and `pull_request_review_comment` at `/webhook/forge/pull-request`, + and one for `workflow_run_failure` and `workflow_run_cancelled` (a finished CI run) at + `/webhook/forge/workflow-run`. At each start it adds + any of these events a registered hook lacks, except for the CI run events: + Forgejo's hook edit never sets them, so the `workflow-run` hook gets them + only when it's created. Adding a run event later means a new hook kind. 2. On each delivery the controller checks the HMAC and publishes the body, unchanged, to the `SWARM_EVENTS` stream on `$SWARM.events.forge.<X-Forgejo-Event-Type>`, such as @@ -78,7 +85,7 @@ 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 three flows above, each a source → match on `action` → +The router ships the six flows above, each a source → match on `action` → todo, 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 @@ -93,7 +100,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, 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` or `workflow_run_cancelled`, and condition and recipient fields from the event's field list. The same goes for the `{field}` placeholders in a todo's key and text. 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`](../../2-handbook/swarm/ui.md#notification-flows) shows the document and @@ -179,6 +186,33 @@ review once the classifier has approved: The skip rule above applies: the router asks for argus's review once per head. +**CI runs as a source.** A finished CI run arrives as `workflow_run_failure` +(a job failed, once every job has ended) or `workflow_run_cancelled`. The body +holds the run, and the run holds the webhook that started it as a JSON string +in `run.event_payload`. For a run on a PR that's the PR's own delivery, so the +router reads the PR from it: `repo`, `number`, `title`, `url` and `author` are +the PR's, `action` is the run's outcome (`failure` or `cancelled`), and +`sender` is the account that triggered the run. A run that started from +anything else, such as a push to a branch with no PR, names no PR and the +router ignores it. The `author` field is the account that opened the issue or +PR, so a flow can send to it: + +| node | setting | +| ------ | ------------------------------------------------------------- | +| source | `workflow_run_failure` | +| todo | `to` = field `author`, key `forge:{repo}#{number}:ci`, upsert | + +A clear of one key, such as `ci-cleared`'s, leaves the agent in the issue's +holders, because it may still hold the assignment todo that `closed` must find. +Only a prefix clear drops the holder. An author whose CI ever failed gets +the `closed` todo when their PR closes or merges, like an assignee. + +A run reports in when it ends, and a push cancels the runs before it, so an old +run can end after the push cleared the `:ci` todo. The router keeps the PR's +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. + **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, diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 85390f6b..df101afb 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -1892,6 +1892,14 @@ fn hook_registration(kind: DeliveryKind) -> (&'static [&'static str], HookScope< ], HookScope::Instance, ), + // Instance-wide too: the outcome of a CI run, which the router tells a + // PR's author about. A hook of its own, because Forgejo's hook edit + // never sets `workflow_run_*` events: a hook can only get them when it + // is created. + DeliveryKind::WorkflowRun => ( + &["workflow_run_failure", "workflow_run_cancelled"], + HookScope::Instance, + ), } } @@ -2244,10 +2252,15 @@ mod tests { "pull_request_review_comment", ] ); + assert_eq!( + events(DeliveryKind::WorkflowRun), + ["workflow_run_failure", "workflow_run_cancelled"] + ); // The events a stored flow's source may name are the ones the hooks // deliver to the router. let mut routed = events(DeliveryKind::Issues); routed.extend(events(DeliveryKind::PullRequest)); + routed.extend(events(DeliveryKind::WorkflowRun)); routed.sort_unstable(); let mut sources = swarm_wire::notify_flows::SOURCE_EVENTS .iter() diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs index bc4b8812..f003782c 100644 --- a/swarm-controller/src/notify_flows.rs +++ b/swarm-controller/src/notify_flows.rs @@ -303,7 +303,7 @@ mod tests { let Json(view) = get_flows(State(state.clone())).await.expect("get"); assert_eq!(view.revision, 0); assert!(view.disabled_builtins.is_empty() && view.flows.is_empty()); - assert_eq!(view.builtins.len(), 4); + assert_eq!(view.builtins.len(), notify_flows::builtins().len()); assert!(view.builtins.iter().any(|f| f.id == "head-classifier")); let json = serde_json::to_value(&view).expect("serializes"); assert_eq!(json["schema"]["limits"]["max_nodes"], 64); @@ -331,7 +331,11 @@ mod tests { .expect("read") .expect("stored"); let doc = Document::decode(&bytes).expect("the router reads it"); - assert_eq!(doc.active().0.len(), 4, "3 built-ins and the custom flow"); + assert_eq!( + doc.active().0.len(), + notify_flows::builtins().len(), + "the built-ins less the switched-off one, and the custom flow" + ); // A refused save changes nothing: an orphan node, and a built-in that does not exist. let mut orphan = custom("orphan"); diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index 587c66c7..7ba56c55 100644 --- a/swarm-controller/src/webhook.rs +++ b/swarm-controller/src/webhook.rs @@ -420,6 +420,13 @@ pub(super) enum DeliveryKind { /// the instance, kept the same way. #[strum(serialize = "pull-request")] PullRequest, + /// Every finished CI run that failed or was cancelled, on every repo in + /// the instance, kept the same way. Its own hook: Forgejo's hook edit + /// ignores the `workflow_run_*` events, so they can only be set when a + /// hook is created, and a `pull-request` hook registered earlier could not + /// gain them. + #[strum(serialize = "workflow-run")] + WorkflowRun, } /// The route prefix a registered `target_url` must point at. @@ -435,13 +442,14 @@ impl DeliveryKind { /// Every kind, for the registration sweep. An array rather than a /// hand-written list at the call site, so adding a kind cannot leave one /// hook unregistered. - pub(super) const ALL: [Self; 6] = [ + pub(super) const ALL: [Self; 7] = [ Self::Knowledge, Self::ConfigPr, Self::VcsActivity, Self::ReviewRequest, Self::Issues, Self::PullRequest, + Self::WorkflowRun, ]; /// The `target_url` to register with Forgejo for this kind, given the @@ -549,8 +557,8 @@ pub(super) fn verify( /// comment for why both this and a periodic poll write the same cache), and a /// merge queues a deploy of the merged commit on the agent's hive. A /// `ReviewRequest` delivery naming the review classifier's account starts a -/// review run in the background. An `Issues` or `PullRequest` delivery is -/// kept in the forge events stream for the notification router +/// review run in the background. An `Issues`, `PullRequest` or `WorkflowRun` +/// delivery is kept in the forge events stream for the notification router /// (`crate::forge_events`). /// /// Returns 200 on an accepted delivery so Forgejo does not retry. A refused @@ -564,7 +572,8 @@ pub(super) fn verify( params( ("kind" = String, Path, description = "which swarm-wide hook this delivery arrived on: \ `knowledge`, `config-pr`, `vcs-activity`, \ - `review-request`, `issues` or `pull-request`") + `review-request`, `issues`, `pull-request` or \ + `workflow-run`") ), request_body( content = String, @@ -637,7 +646,7 @@ pub(super) async fn post_webhook_forge( } } DeliveryKind::ReviewRequest => crate::review_classifier::review::on_delivery(&state, &body), - DeliveryKind::Issues | DeliveryKind::PullRequest => { + DeliveryKind::Issues | DeliveryKind::PullRequest | DeliveryKind::WorkflowRun => { 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(); diff --git a/swarm-notify-router/src/dry_run.rs b/swarm-notify-router/src/dry_run.rs index 5a5775e2..752806b2 100644 --- a/swarm-notify-router/src/dry_run.rs +++ b/swarm-notify-router/src/dry_run.rs @@ -24,7 +24,7 @@ pub fn run(flows: &[Flow], events: &[(String, Vec<u8>)]) -> DryRun { let mut snapshots: BTreeMap<String, Snapshot> = BTreeMap::new(); let mut run = 0; for (event_type, body) in events { - let Some(delivery) = Delivery::parse(body) else { + let Some(delivery) = Delivery::from_event(event_type, body) else { continue; }; run += 1; diff --git a/swarm-notify-router/src/event.rs b/swarm-notify-router/src/event.rs index f446a440..1bb79833 100644 --- a/swarm-notify-router/src/event.rs +++ b/swarm-notify-router/src/event.rs @@ -26,6 +26,11 @@ pub struct Delivery { pub title: String, pub url: String, pub sender: String, + /// The account that opened the issue or PR. + pub author: String, + /// The commit at the PR's head when the delivery was made; empty for an + /// issue. A CI run's delivery carries the PR as it was when the run began. + pub head: String, pub is_pull: bool, pub assignees: BTreeSet<String>, /// The PR is a draft: the forge's `pull_request.draft`, true for a title @@ -82,6 +87,15 @@ impl Delivery { .get("sender") .map(|s| text(s, "login")) .unwrap_or_default(), + author: item + .get("user") + .map(|u| text(u, "login")) + .unwrap_or_default(), + head: item + .pointer("/head/sha") + .and_then(Json::as_str) + .unwrap_or_default() + .to_owned(), is_pull, assignees, draft, @@ -89,6 +103,30 @@ impl Delivery { }) } + /// Read the delivery of Forgejo event type `event_type`. A `workflow_run_*` + /// delivery carries the webhook that started the run as a JSON string in + /// `run.event_payload`; for a run on a PR that is the PR's own delivery, so + /// the PR is read from it, `action` is the run's outcome (`failure`, + /// `cancelled`) and `sender` is the account that triggered the run. A run + /// on anything else (a push to a branch with no PR) names no PR and gives + /// `None`. + pub fn from_event(event_type: &str, body: &[u8]) -> Option<Self> { + let Some(outcome) = event_type.strip_prefix("workflow_run_") else { + return Self::parse(body); + }; + let json: Json = serde_json::from_slice(body).ok()?; + let run = json.get("run")?; + let mut delivery = Self::parse(run.get("event_payload")?.as_str()?.as_bytes())?; + if !delivery.is_pull { + return None; + } + outcome.clone_into(&mut delivery.action); + if let Some(trigger) = run.pointer("/trigger_user/login").and_then(Json::as_str) { + trigger.clone_into(&mut delivery.sender); + } + Some(delivery) + } + /// The state bucket key for this delivery's issue: `<repo id>.<number>`, /// numeric so it is always a valid key whatever the repo is named. pub fn state_key(&self) -> String { @@ -122,6 +160,11 @@ pub struct Snapshot { /// The agents holding a router todo for this issue. #[serde(default)] pub holders: BTreeSet<String>, + /// The commit at the head of the PR as of the last delivery about the PR + /// that was no CI run. A run that began at another commit is superseded + /// and tells nobody anything. + #[serde(default)] + pub head: Option<String>, } /// Whether `event_type` is an assign delivery, whose assignee list is the @@ -153,6 +196,7 @@ pub fn fields(event_type: &str, delivery: &Delivery, snapshot: &Snapshot) -> Fie ("title", text(&delivery.title)), ("url", text(&delivery.url)), ("sender", text(&delivery.sender)), + ("author", text(&delivery.author)), ("draft", text(bool_text(delivery.draft))), ("undrafted", text(bool_text(delivery.undrafted))), ( @@ -266,6 +310,89 @@ mod tests { serde_json::to_vec(&body).expect("encodes") } + /// A `workflow_run_*` delivery: the run, holding the webhook that started + /// it as a JSON string. + fn run_body(started_by: &Json) -> Vec<u8> { + serde_json::to_vec(&serde_json::json!({ + "action": "failure", + "run": { + "id": 5, + "trigger_user": { "login": "atlas" }, + "event_payload": serde_json::to_string(started_by).expect("encodes"), + }, + })) + .expect("encodes") + } + + fn pull_payload(author: &str) -> Json { + serde_json::json!({ + "action": "synchronized", + "number": 7, + "pull_request": { + "number": 7, "title": "t", "html_url": "u", "user": { "login": author }, + "head": { "sha": "abc123" }, + }, + "repository": { "id": 1, "full_name": "o/r" }, + "sender": { "login": "pusher" }, + }) + } + + #[test] + fn a_run_on_a_pull_request_reads_as_that_pull_request_with_the_runs_outcome() { + let body = run_body(&pull_payload("iris")); + for (event, outcome) in [ + ("workflow_run_failure", "failure"), + ("workflow_run_cancelled", "cancelled"), + ] { + let d = Delivery::from_event(event, &body).expect("a PR run"); + assert!(d.is_pull); + assert_eq!((d.repo.as_str(), d.number), ("o/r", 7)); + assert_eq!(d.action, outcome); + assert_eq!(d.author, "iris"); + assert_eq!(d.head, "abc123", "the commit the run began at"); + assert_eq!(d.sender, "atlas", "the account that triggered the run"); + assert_eq!(d.state_key(), "1.7"); + let f = fields(event, &d, &Snapshot::default()); + assert_eq!(f.get("author"), Some(&Value::Text("iris".to_owned()))); + assert_eq!(f.get("kind"), Some(&Value::Text("PR".to_owned()))); + } + } + + #[test] + fn a_run_that_names_no_pull_request_is_ignored() { + let push = serde_json::json!({ + "ref": "refs/heads/main", + "repository": { "id": 1, "full_name": "o/r" }, + }); + assert_eq!( + Delivery::from_event("workflow_run_failure", &run_body(&push)), + None + ); + let issue: Json = + serde_json::from_slice(&issue_assign("assigned", &["atlas"])).expect("json"); + assert_eq!( + Delivery::from_event("workflow_run_failure", &run_body(&issue)), + None + ); + assert_eq!( + Delivery::from_event("workflow_run_failure", b"not json"), + None + ); + assert_eq!( + Delivery::from_event("workflow_run_failure", b"{\"run\":{}}"), + None + ); + } + + #[test] + fn other_events_read_the_body_directly() { + let body = pull_body("opened", "t", false, None); + assert_eq!( + Delivery::from_event("pull_request", &body), + Delivery::parse(&body) + ); + } + #[test] fn the_forges_draft_flag_marks_a_draft() { let d = Delivery::parse(&pull_body("synchronized", "WIP: t", true, None)).expect("parses"); @@ -307,6 +434,7 @@ mod tests { let before = Snapshot { assignees: Some(BTreeSet::from(["atlas".to_owned(), "iris".to_owned()])), holders: BTreeSet::new(), + ..Snapshot::default() }; let f = fields("issue_assign", &d, &before); assert_eq!(list(&f, "assignees_removed"), ["atlas"]); @@ -316,6 +444,7 @@ mod tests { let before = Snapshot { assignees: Some(BTreeSet::from(["iris".to_owned()])), holders: BTreeSet::new(), + ..Snapshot::default() }; assert_eq!( list(&fields("issue_assign", &d, &before), "assignees_added"), diff --git a/swarm-notify-router/src/main.rs b/swarm-notify-router/src/main.rs index cdf73e4b..61bcac65 100644 --- a/swarm-notify-router/src/main.rs +++ b/swarm-notify-router/src/main.rs @@ -216,7 +216,7 @@ impl Router { let event_type = forge_events::event_type(&message.subject) .ok_or_else(|| Skip::Unusable("not a forge event subject".to_owned()))? .to_owned(); - let Some(delivery) = event::Delivery::parse(&message.payload) else { + let Some(delivery) = event::Delivery::from_event(&event_type, &message.payload) else { // A delivery that names no issue or PR has nothing to route. return Ok(Handled::Ignored); }; diff --git a/swarm-notify-router/src/route.rs b/swarm-notify-router/src/route.rs index 760f8b92..c5765357 100644 --- a/swarm-notify-router/src/route.rs +++ b/swarm-notify-router/src/route.rs @@ -32,6 +32,21 @@ pub fn route( flows: &[Flow], is_agent: impl Fn(&str) -> bool, ) -> Routed { + let is_run = event_type.starts_with("workflow_run_"); + // Forgejo cancels the runs of a PR when it is pushed to, and an old run can + // finish after the push cleared the PR's CI todo. Such a run began at a + // commit the PR has since moved past, so it tells nobody anything. + if is_run + && !delivery.head.is_empty() + && snapshot.head.as_deref().is_some_and(|h| h != delivery.head) + { + return Routed { + actions: Vec::new(), + requests: Vec::new(), + skipped: Vec::new(), + next: snapshot.clone(), + }; + } let fields = fields(event_type, delivery, snapshot); let mut actions = Vec::new(); let mut skipped = Vec::new(); @@ -47,6 +62,11 @@ pub fn route( } } let mut next = snapshot.clone(); + // A run's delivery carries the PR as it was when the run began, so only + // the other deliveries say where the PR's head is now. + if !is_run && delivery.is_pull && !delivery.head.is_empty() { + next.head = Some(delivery.head.clone()); + } 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 @@ -61,9 +81,13 @@ pub fn route( Op::Upsert => { next.holders.insert(action.agent.clone()); } - Op::Clear => { + // Only a prefix clear can take every todo the agent held for the + // issue. A clear of one key leaves it a holder: it may still hold + // the assignment, which the `closed` flow must find. + Op::Clear if action.op.prefix => { cleared.insert(action.agent.clone()); } + Op::Clear => {} } } next.holders.retain(|h| !cleared.contains(h)); @@ -89,6 +113,8 @@ mod tests { title: "t".to_owned(), url: "u".to_owned(), sender: "mara".to_owned(), + author: "mara".to_owned(), + head: String::new(), is_pull: false, assignees: assignees.iter().map(|a| (*a).to_owned()).collect(), draft: false, @@ -103,6 +129,135 @@ mod tests { .collect() } + fn pull_by(author: &str, action: &str) -> Delivery { + Delivery { + is_pull: true, + author: author.to_owned(), + ..delivery(action, &[]) + } + } + + /// A failed run tells the PR's author, a cancelled one rewrites that todo, + /// and a push clears it. The author keeps holding the PR, so a later close + /// still reaches them. + #[test] + fn a_failed_run_tells_the_author_and_a_push_clears_it() { + let flows = builtins(); + let is_agent = |a: &str| a == "atlas"; + + let failed = route( + "workflow_run_failure", + &pull_by("atlas", "failure"), + &Snapshot::default(), + &flows, + is_agent, + ); + assert_eq!(agents(&failed), [("atlas", Op::Upsert)]); + assert_eq!(failed.actions[0].op.key, "forge:o/r#9:ci"); + assert!( + failed.actions[0] + .op + .summary + .starts_with("CI failed on PR o/r#9") + ); + assert_eq!(failed.next.holders, BTreeSet::from(["atlas".to_owned()])); + + let cancelled = route( + "workflow_run_cancelled", + &pull_by("atlas", "cancelled"), + &failed.next, + &flows, + is_agent, + ); + assert_eq!(agents(&cancelled), [("atlas", Op::Upsert)]); + assert_eq!(cancelled.actions[0].op.key, "forge:o/r#9:ci"); + assert!( + cancelled.actions[0] + .op + .summary + .starts_with("CI was cancelled on PR") + ); + + let pushed = route( + "pull_request_sync", + &pull_by("atlas", "synchronized"), + &cancelled.next, + &flows, + is_agent, + ); + let ci_clear: Vec<_> = pushed + .actions + .iter() + .filter(|a| a.op.op == Op::Clear) + .collect(); + assert_eq!(ci_clear.len(), 1); + assert_eq!(ci_clear[0].op.key, "forge:o/r#9:ci"); + assert!( + !ci_clear[0].op.prefix, + "only the CI todo goes, not the assignment" + ); + assert_eq!(pushed.next.holders, BTreeSet::from(["atlas".to_owned()])); + } + + fn at_head(mut d: Delivery, head: &str) -> Delivery { + head.clone_into(&mut d.head); + d + } + + /// 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. + #[test] + fn a_run_from_before_the_latest_push_tells_nobody() { + let flows = builtins(); + let is_agent = |a: &str| a == "atlas"; + let run = |event: &str, head: &str, before: &Snapshot| { + // The router names a run's action after its outcome. + let outcome = event.trim_start_matches("workflow_run_"); + route( + event, + &at_head(pull_by("atlas", outcome), head), + before, + &flows, + is_agent, + ) + }; + + // Before the router has seen the PR's head it can't tell, so it tells. + let unknown = run("workflow_run_cancelled", "old", &Snapshot::default()); + assert_eq!(agents(&unknown), [("atlas", Op::Upsert)]); + assert_eq!(unknown.next.head, None, "a run says nothing about the head"); + + let pushed = route( + "pull_request_sync", + &at_head(pull_by("atlas", "synchronized"), "new"), + &unknown.next, + &flows, + is_agent, + ); + assert_eq!(pushed.next.head.as_deref(), Some("new")); + + let late = run("workflow_run_cancelled", "old", &pushed.next); + assert!(late.actions.is_empty() && late.skipped.is_empty()); + assert_eq!(late.next, pushed.next); + + let current = run("workflow_run_failure", "new", &pushed.next); + assert_eq!(agents(¤t), [("atlas", Op::Upsert)]); + } + + #[test] + fn a_failed_run_on_a_pull_request_by_a_non_agent_queues_nothing() { + let got = route( + "workflow_run_failure", + &pull_by("mara", "failure"), + &Snapshot::default(), + &builtins(), + |a| a == "atlas", + ); + assert!(got.actions.is_empty()); + assert_eq!(got.skipped, ["mara"]); + } + /// Assign, reassign, close: the three built-ins in the order an issue /// meets them, carrying the snapshot from one event to the next. #[test] @@ -208,6 +363,7 @@ mod tests { let before = Snapshot { assignees: Some(BTreeSet::from(["atlas".to_owned()])), holders: BTreeSet::from(["atlas".to_owned()]), + ..Snapshot::default() }; let r = route("pull_request_sync", &pull, &before, &flows, |_| false); assert_eq!( @@ -220,7 +376,10 @@ mod tests { reviewer: "review-classifier".to_owned(), }] ); - assert!(r.actions.is_empty() && r.skipped.is_empty()); + // The push also clears the author's CI todo, but the author is no + // agent here, so that goes nowhere. + assert!(r.actions.is_empty()); + assert_eq!(r.skipped, ["mara"]); assert_eq!(r.next, before); // The source gates: another PR event requests nothing. @@ -271,6 +430,7 @@ mod tests { let before = Snapshot { assignees: Some(BTreeSet::from(["atlas".to_owned()])), holders: BTreeSet::from(["atlas".to_owned()]), + ..Snapshot::default() }; let r = route( "issue_comment", diff --git a/swarm-notify-router/src/stored.rs b/swarm-notify-router/src/stored.rs index af3993f5..ec8bbc38 100644 --- a/swarm-notify-router/src/stored.rs +++ b/swarm-notify-router/src/stored.rs @@ -149,6 +149,17 @@ mod tests { use super::*; + /// The built-in flows, in the order the router runs them. + const BUILTIN_IDS: [&str; 7] = [ + "assigned", + "unassigned", + "closed", + "head-classifier", + "ci-failed", + "ci-cancelled", + "ci-cleared", + ]; + fn ids(f: &Flows) -> Vec<&str> { f.active.iter().map(|f| f.id.as_str()).collect() } @@ -188,28 +199,30 @@ mod tests { #[test] fn with_no_document_the_router_runs_the_builtins() { let mut f = Flows::builtins(); - assert_eq!( - ids(&f), - ["assigned", "unassigned", "closed", "head-classifier"] - ); + assert_eq!(ids(&f), BUILTIN_IDS); f.apply(Read::Absent); - assert_eq!( - ids(&f), - ["assigned", "unassigned", "closed", "head-classifier"] - ); + assert_eq!(ids(&f), BUILTIN_IDS); // An empty document is the same. f.apply(at(1, doc(&[], &[]))); - assert_eq!( - ids(&f), - ["assigned", "unassigned", "closed", "head-classifier"] - ); + assert_eq!(ids(&f), BUILTIN_IDS); } #[test] fn a_document_switches_builtins_off_and_adds_custom_flows() { let mut f = Flows::builtins(); f.apply(at(1, doc(&["unassigned"], &["triage"]))); - assert_eq!(ids(&f), ["assigned", "closed", "head-classifier", "triage"]); + assert_eq!( + ids(&f), + [ + "assigned", + "closed", + "head-classifier", + "ci-failed", + "ci-cancelled", + "ci-cleared", + "triage" + ] + ); } #[test] @@ -217,7 +230,10 @@ mod tests { let mut f = Flows::builtins(); f.apply(at(1, doc(&["closed"], &["triage"]))); let before = ids(&f).join(","); - assert_eq!(before, "assigned,unassigned,head-classifier,triage"); + assert_eq!( + before, + "assigned,unassigned,head-classifier,ci-failed,ci-cancelled,ci-cleared,triage" + ); for bad in [ junk(2, b"not json"), junk(3, br#"{"version":2,"flows":[]}"#), @@ -230,20 +246,14 @@ mod tests { // The next good revision replaces them: the control that the bad // ones were kept out by being bad, not by the loader being stuck. f.apply(at(5, doc(&[], &[]))); - assert_eq!( - ids(&f), - ["assigned", "unassigned", "closed", "head-classifier"] - ); + assert_eq!(ids(&f), BUILTIN_IDS); } #[test] fn a_bad_first_document_is_the_builtins_not_nothing() { let mut f = Flows::builtins(); f.apply(junk(1, br#"{"version":9}"#)); - assert_eq!( - ids(&f), - ["assigned", "unassigned", "closed", "head-classifier"] - ); + assert_eq!(ids(&f), BUILTIN_IDS); } #[test] @@ -251,10 +261,7 @@ mod tests { let mut f = Flows::builtins(); f.apply(at(1, doc(&["closed"], &["triage"]))); f.apply(Read::Absent); - assert_eq!( - ids(&f), - ["assigned", "unassigned", "closed", "head-classifier"] - ); + assert_eq!(ids(&f), BUILTIN_IDS); } #[test] @@ -272,9 +279,6 @@ mod tests { revision: 1, bytes: serde_json::to_vec(&d).unwrap(), }); - assert_eq!( - ids(&f), - ["assigned", "unassigned", "closed", "head-classifier"] - ); + assert_eq!(ids(&f), BUILTIN_IDS); } } diff --git a/swarm-wire/src/notify_flows.rs b/swarm-wire/src/notify_flows.rs index 0d634349..40c45137 100644 --- a/swarm-wire/src/notify_flows.rs +++ b/swarm-wire/src/notify_flows.rs @@ -74,6 +74,14 @@ pub const SOURCE_EVENTS: &[(&str, &str)] = &[ "pull_request_review_comment", "A review submitted on a pull request that only comments", ), + ( + "workflow_run_failure", + "A CI run on a pull request that finished with a failed job", + ), + ( + "workflow_run_cancelled", + "A CI run on a pull request that was cancelled", + ), ]; /// The fields of an event a condition or a recipient can name, with what each @@ -87,6 +95,10 @@ pub const FIELDS: &[(&str, &str)] = &[ ("title", "The issue or pull request title"), ("url", "The web link to the issue or pull request"), ("sender", "The account that caused the event"), + ( + "author", + "The account that opened the issue or pull request", + ), ( "draft", "true when the pull request is a draft (its title starts WIP: or [WIP]), else false", @@ -781,10 +793,15 @@ impl TodoSink { /// `closed` key. /// - `head-classifier`: a PR opened, reopened or pushed to requests the review /// classifier's review. +/// - `ci-failed` and `ci-cancelled`: a CI run on a PR that failed or was +/// cancelled gives the PR's author a todo. Both write the same key, so the +/// 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. #[must_use] pub fn builtins() -> Vec<Flow> { let assignment_key = "forge:{repo}#{number}:assignment"; - vec![ + let mut flows = vec![ chain( "assigned", &["issue_assign", "pull_request_assign"], @@ -812,6 +829,47 @@ pub fn builtins() -> Vec<Flow> { ), closed(), head_classifier(), + ]; + flows.extend(ci_flows()); + flows +} + +/// 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> { + let ci_key = "forge:{repo}#{number}:ci"; + let to_author = |op, text: &str| TodoSink { + to: Recipients::Field("author".to_owned()), + op, + key: ci_key.to_owned(), + prefix: false, + text: text.to_owned(), + }; + vec![ + chain( + "ci-failed", + &["workflow_run_failure"], + "failure", + to_author( + Op::Upsert, + "CI failed on {kind} {repo}#{number}: {title} ({url})", + ), + ), + chain( + "ci-cancelled", + &["workflow_run_cancelled"], + "cancelled", + to_author( + Op::Upsert, + "CI was cancelled on {kind} {repo}#{number}: {title} ({url})", + ), + ), + chain( + "ci-cleared", + &["pull_request_sync"], + "synchronized", + to_author(Op::Clear, ""), + ), ] } @@ -1385,7 +1443,15 @@ mod tests { let ids = |d: &Document| d.active().0.into_iter().map(|f| f.id).collect::<Vec<_>>(); assert_eq!( ids(&Document::default()), - ["assigned", "unassigned", "closed", "head-classifier"], + [ + "assigned", + "unassigned", + "closed", + "head-classifier", + "ci-failed", + "ci-cancelled", + "ci-cleared" + ], "an empty document is the built-ins" ); let mut off = custom("off"); @@ -1399,7 +1465,15 @@ mod tests { }; assert_eq!( ids(&doc), - ["assigned", "closed", "head-classifier", "triage"] + [ + "assigned", + "closed", + "head-classifier", + "ci-failed", + "ci-cancelled", + "ci-cleared", + "triage" + ] ); let (_, skipped) = doc.active(); assert_eq!(skipped.len(), 2, "{skipped:?}");