From e8538d8690281d37469b329a86112b42bdc8309d Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 25 Oct 2026 01:51:16 +0200 Subject: [PATCH] swarm-controller: POST /api/notify/dry-run runs a draft flows document --- docs/swarm/notify-router.md | 17 +- swarm-controller/src/notify_flows.rs | 21 +- swarm-controller/src/notify_router_api.rs | 222 +++++++++++++++++++++- swarm-controller/src/startup.rs | 5 +- 4 files changed, 244 insertions(+), 21 deletions(-) diff --git a/docs/swarm/notify-router.md b/docs/swarm/notify-router.md index f4ec6f71..50d9ce5a 100644 --- a/docs/swarm/notify-router.md +++ b/docs/swarm/notify-router.md @@ -204,14 +204,17 @@ counts as an agent. **The API.** The router answers two NATS requests from swarm-controller, which serves them next to `GET /api/notify/flows`: -| Route | Router subject | Answers | -| ---------------------------- | ------------------------------ | ------------------------------------------------------------------ | -| `GET /api/notify/deliveries` | `$SWARM.router.api.deliveries` | the newest rows, newest first; `?limit=` up to 500, 100 by default | -| `GET /api/notify/dry-run` | `$SWARM.router.api.dry-run` | the saved flows document, run over the newest 50 events | +| Route | Router subject | Answers | +| ---------------------------- | ------------------------------ | -------------------------------------------------------------------------------------------- | +| `GET /api/notify/deliveries` | `$SWARM.router.api.deliveries` | the newest rows, newest first; `?limit=` up to 500, 100 by default | +| `GET /api/notify/dry-run` | `$SWARM.router.api.dry-run` | the saved flows document, run over the newest 50 events | +| `POST /api/notify/dry-run` | `$SWARM.router.api.dry-run` | the body's draft flows document, checked as a `PUT` checks it, run over the newest 50 events | -The dry-run request carries a whole document, so a draft runs the same way. -The controller sends the saved one. [`/notifications`](ui.md#notification-flows) -shows both: the recent deliveries, and the counts beside each node. +The dry-run request carries a whole document. The `GET` sends the saved one; +the `POST` takes a `PUT` body (it ignores a `revision`), answers 400 +with the `PUT` error for a document that fails the check, and forwards a valid one. +[`/notifications`](ui.md#notification-flows) shows both: the recent deliveries, +and the counts beside each node. **Deploying in any order.** A router that predates these subjects doesn't subscribe to them, and the queue answers the controller's request at once with diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs index 7cd4b4e1..bc4b8812 100644 --- a/swarm-controller/src/notify_flows.rs +++ b/swarm-controller/src/notify_flows.rs @@ -165,6 +165,20 @@ pub(crate) fn unreadable(why: &DecodeError) -> ProblemDetails { ) } +/// The document a body describes, or the first way it fails +/// [`Document::check`]. +pub(crate) fn checked_document( + disabled_builtins: Vec, + flows: Vec, +) -> Result { + let doc = Document { + version: notify_flows::VERSION, + disabled_builtins, + flows, + }; + doc.check().map(|()| doc) +} + /// Replace the flows. #[utoipa::path( put, @@ -196,12 +210,7 @@ pub(crate) async fn put_flows( body: Result, JsonRejection>, ) -> Result, ProblemDetails> { let Json(put) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; - let doc = Document { - version: notify_flows::VERSION, - disabled_builtins: put.disabled_builtins, - flows: put.flows, - }; - doc.check() + let doc = checked_document(put.disabled_builtins, put.flows) .map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?; let client = queue(&state).map_err(refused)?; let store = open(&client).await.map_err(|e| queue_problem(&e))?; diff --git a/swarm-controller/src/notify_router_api.rs b/swarm-controller/src/notify_router_api.rs index c6cbfa04..2922df9e 100644 --- a/swarm-controller/src/notify_router_api.rs +++ b/swarm-controller/src/notify_router_api.rs @@ -1,28 +1,32 @@ -//! `GET /api/notify/deliveries` and `GET /api/notify/dry-run`: what the -//! notification router has delivered, and what the saved flows would do with +//! `GET /api/notify/deliveries`, `GET /api/notify/dry-run` and +//! `POST /api/notify/dry-run`: what the +//! notification router has delivered, and what the saved flows, or a draft, +//! would do with //! the newest forge events. //! //! The router keeps the log and the events, so each is a NATS request to it //! (`swarm_wire::router_api`). A router that predates them, or none at all, //! has no subscriber: the queue answers "no responders" at once and these //! answer 503 saying so, as they do when it hears the request and stays -//! silent. The saved document is what `GET /api/notify/flows` shows; a draft -//! is run by sending the same request with another document. +//! silent. The saved document is what `GET /api/notify/flows` shows; the `POST` +//! sends the same request with the body's document, checked as a `PUT` checks +//! it. use axum::Json; +use axum::extract::rejection::JsonRejection; use axum::extract::{Query, State}; use axum::http::StatusCode; use problem_details::ProblemDetails; use serde::Deserialize; use serde::de::DeserializeOwned; -use swarm_wire::notify_flows::Document; +use swarm_wire::notify_flows::{Document, Flow}; use swarm_wire::router_api::{ DELIVERIES_SUBJECT, DRY_RUN_SUBJECT, Deliveries, DeliveriesRequest, DryRun, DryRunRequest, REPLY_DEADLINE, Refused, }; -use utoipa::IntoParams; +use utoipa::{IntoParams, ToSchema}; -use crate::notify_flows::{Stored, open, queue_problem, stored, unreadable}; +use crate::notify_flows::{Stored, checked_document, open, queue_problem, stored, unreadable}; use crate::routes::{AppState, error_problem}; use crate::schedules::{queue, refused}; @@ -136,8 +140,56 @@ pub(crate) async fn get_dry_run( ask(&client, DRY_RUN_SUBJECT, body).await.map(Json) } +/// The body of `POST /api/notify/dry-run`: the document of a `PUT`, whose +/// `revision` may be present and is ignored. +#[derive(Debug, Clone, Deserialize, ToSchema)] +#[serde(deny_unknown_fields)] +pub(crate) struct DraftFlows { + /// Ignored. + #[serde(default, rename = "revision")] + _revision: u64, + /// The ids of the built-in flows to switch off. + #[serde(default)] + disabled_builtins: Vec, + /// The custom flows. + #[serde(default)] + flows: Vec, +} + +/// A draft flows document run over the newest 50 forge events, delivering and +/// saving nothing: how many events reached each node. +#[utoipa::path( + post, + path = "/api/notify/dry-run", + request_body = DraftFlows, + responses( + (status = 200, description = "per flow that would run, the number of the newest 50 \ + events that reached each node", body = DryRun), + (status = 400, description = "not JSON or not this shape, or the document fails the \ + check `PUT /api/notify/flows` applies (problem+json)", body = String), + (status = 502, description = "the router refused, or its answer did not decode \ + (problem+json)", body = String), + (status = 503, description = "no swarm queue is configured or connected here, or the \ + router does not answer: it is not running, or it predates the dry run \ + (problem+json)", body = String), + ), + tag = "notify" +)] +pub(crate) async fn post_dry_run( + State(state): State, + body: Result, JsonRejection>, +) -> Result, ProblemDetails> { + let Json(draft) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; + let document = checked_document(draft.disabled_builtins, draft.flows) + .map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?; + let client = queue(&state).map_err(refused)?; + let body = serde_json::to_vec(&DryRunRequest { document }).expect("a request encodes"); + ask(&client, DRY_RUN_SUBJECT, body).await.map(Json) +} + #[cfg(test)] mod tests { + use axum::extract::rejection::MissingJsonContentType; use futures_util::StreamExt as _; use swarm_wire::notify_flows::{self, Document}; use swarm_wire::router_api::{FlowHits, NodeHits}; @@ -332,4 +384,160 @@ mod tests { let unreadable = get_dry_run(State(state)).await.expect_err("version 2"); assert_eq!(unreadable.status, Some(StatusCode::CONFLICT)); } + + fn draft(flows: &serde_json::Value) -> Json { + Json( + serde_json::from_value(serde_json::json!({ + "revision": 7, + "disabled_builtins": ["closed"], + "flows": flows, + })) + .expect("a draft"), + ) + } + + fn sink_flow(id: &str) -> serde_json::Value { + serde_json::json!({ + "id": id, "name": "n", "enabled": true, + "nodes": [ + {"id": "source", "kind": "source", "events": ["issues"]}, + {"id": "sink", "kind": "todo", "to": {"agent": "triage"}, "op": "upsert", + "key": "k", "prefix": false, "text": "t"}, + ], + "edges": [{"from": "source", "to": "sink"}], + }) + } + + /// A valid draft reaches the router as the request's document, with the + /// body's `revision` dropped, and the router's counts come back. + #[tokio::test] + async fn a_valid_draft_is_forwarded_and_the_counts_come_back() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::state(&client); + let reply = DryRun { + events: 2, + flows: vec![FlowHits { + flow: "mine".to_owned(), + nodes: vec![NodeHits { + node: "source".to_owned(), + hits: 2, + }], + }], + }; + let router = answer( + &client, + DRY_RUN_SUBJECT, + serde_json::to_vec(&reply).unwrap(), + ) + .await; + let Json(got) = post_dry_run( + State(state), + Ok(draft(&serde_json::json!([sink_flow("mine")]))), + ) + .await + .expect("answered"); + assert_eq!(got, reply); + let sent: DryRunRequest = + serde_json::from_slice(&router.await.expect("task")).expect("a request"); + assert_eq!(sent.document.disabled_builtins, ["closed"]); + assert_eq!(sent.document.flows.len(), 1); + assert_eq!(sent.document.flows[0].id, "mine"); + } + + /// A tree that fails the check is 400 with `PUT`'s message and never + /// reaches the router; a body that is not a document is 400 too. The + /// control is the valid draft above, which the same router subscription + /// would have received. + #[tokio::test] + async fn an_invalid_draft_is_400_and_never_forwarded() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::state(&client); + let mut requests = client.subscribe(DRY_RUN_SUBJECT).await.expect("subscribed"); + client.flush().await.expect("flushed"); + + let mut orphan = sink_flow("orphan"); + orphan["edges"] = serde_json::json!([]); + let refused = post_dry_run( + State(state.clone()), + Ok(draft(&serde_json::json!([orphan]))), + ) + .await + .expect_err("an orphan node"); + assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST)); + let put = checked_document( + Vec::new(), + serde_json::from_value(serde_json::json!([{ + "id": "orphan", "name": "n", "enabled": true, + "nodes": [ + {"id": "source", "kind": "source", "events": ["issues"]}, + {"id": "sink", "kind": "todo", "to": {"agent": "triage"}, "op": "upsert", + "key": "k", "prefix": false, "text": "t"}, + ], + "edges": [], + }])) + .unwrap(), + ) + .expect_err("the same check"); + assert_eq!( + refused.detail.as_deref(), + Some(put.as_str()), + "PUT's message" + ); + + let typo: Result, JsonRejection> = Err( + JsonRejection::MissingJsonContentType(MissingJsonContentType::default()), + ); + let refused = post_dry_run(State(state.clone()), typo) + .await + .expect_err("not json"); + assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST)); + + // Nothing was forwarded: the next request on the subject is the + // control's, not either refusal's. + let control = post_dry_run( + State(state), + Ok(draft(&serde_json::json!([sink_flow("ok")]))), + ); + let (sent, got) = tokio::join!( + async { + let m = requests.next().await.expect("the control's request"); + let sent: DryRunRequest = serde_json::from_slice(&m.payload).expect("request"); + client + .publish( + m.reply.expect("reply"), + serde_json::to_vec(&DryRun { + events: 0, + flows: Vec::new(), + }) + .unwrap() + .into(), + ) + .await + .expect("answered"); + sent + }, + control + ); + assert_eq!(sent.document.flows[0].id, "ok"); + assert!(got.is_ok()); + } + + /// With no router the draft is 503; with no queue it is 503 too. + #[tokio::test] + async fn a_draft_without_a_router_is_unavailable() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::state(&client); + let b = post_dry_run(State(state), Ok(draft(&serde_json::json!([])))) + .await + .expect_err("nobody answers"); + assert_eq!(b.status, Some(StatusCode::SERVICE_UNAVAILABLE)); + let (state, _) = crate::routes::tests::state_with_roster(); + let b = post_dry_run(State(state), Ok(draft(&serde_json::json!([])))) + .await + .expect_err("no queue"); + assert_eq!(b.status, Some(StatusCode::SERVICE_UNAVAILABLE)); + } } diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index a27bbd7f..3e8de038 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -645,7 +645,10 @@ pub(crate) fn build_app(state: AppState) -> axum::Router { )) .routes(routes!(notify_flows::get_flows, notify_flows::put_flows)) .routes(routes!(notify_router_api::get_deliveries)) - .routes(routes!(notify_router_api::get_dry_run)) + .routes(routes!( + notify_router_api::get_dry_run, + notify_router_api::post_dry_run + )) .routes(routes!(webhook::post_webhook_forge)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from