vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-controller: POST /api/notify/dry-run runs a draft flows document

Browse source
This commit is contained in:
atlas 2026-10-25 01:51:16 +02:00 • committed by mara
commit e8538d8690
4 changed files with 245 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

17
docs/swarm/notify-router.md
View file

@ -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

21
swarm-controller/src/notify_flows.rs
View file

@ -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<String>,
flows: Vec<Flow>,
) -> Result<Document, String> {
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<Json<PutFlows>, JsonRejection>,
) -> Result<Json<FlowsView>, 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))?;

222
swarm-controller/src/notify_router_api.rs
View file

@ -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<String>,
/// The custom flows.
#[serde(default)]
flows: Vec<Flow>,
}
/// 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<AppState>,
body: Result<Json<DraftFlows>, JsonRejection>,
) -> Result<Json<DryRun>, 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<DraftFlows> {
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<Json<DraftFlows>, 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));
}
}

5
swarm-controller/src/startup.rs
View file

@ -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