543 lines
21 KiB
Rust
543 lines
21 KiB
Rust
//! `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; 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, Flow};
|
|
use swarm_wire::router_api::{
|
|
DELIVERIES_SUBJECT, DRY_RUN_SUBJECT, Deliveries, DeliveriesRequest, DryRun, DryRunRequest,
|
|
REPLY_DEADLINE, Refused,
|
|
};
|
|
use utoipa::{IntoParams, ToSchema};
|
|
|
|
use crate::notify_flows::{Stored, checked_document, open, queue_problem, stored, unreadable};
|
|
use crate::routes::{AppState, error_problem};
|
|
use crate::schedules::{queue, refused};
|
|
|
|
/// Ask the router on `subject` and decode its answer as `T`.
|
|
///
|
|
/// 503: nothing hears the request, or what hears it does not answer in time;
|
|
/// both mean the running router does not have this feature. 502: it answered
|
|
/// with a refusal or with something this build cannot read.
|
|
async fn ask<T: DeserializeOwned>(
|
|
client: &async_nats::Client,
|
|
subject: &str,
|
|
body: Vec<u8>,
|
|
) -> Result<T, ProblemDetails> {
|
|
use async_nats::RequestErrorKind;
|
|
|
|
let sent = async_nats::Request::new()
|
|
.payload(body.into())
|
|
.timeout(Some(REPLY_DEADLINE));
|
|
let reply = client
|
|
.send_request(subject.to_owned(), sent)
|
|
.await
|
|
.map_err(|e| {
|
|
let detail = match e.kind() {
|
|
RequestErrorKind::NoResponders => "the notification router does not answer this: \
|
|
it is not running, or it is an older build than this controller"
|
|
.to_owned(),
|
|
RequestErrorKind::TimedOut => format!(
|
|
"the notification router sent no answer within {}s",
|
|
REPLY_DEADLINE.as_secs()
|
|
),
|
|
_ => format!("asking the notification router failed: {e}"),
|
|
};
|
|
error_problem(StatusCode::SERVICE_UNAVAILABLE, &detail)
|
|
})?;
|
|
if let Ok(answer) = serde_json::from_slice::<T>(&reply.payload) {
|
|
return Ok(answer);
|
|
}
|
|
let detail = match serde_json::from_slice::<Refused>(&reply.payload) {
|
|
Ok(Refused { error }) => format!("the notification router refused: {error}"),
|
|
Err(e) => format!("the notification router's answer did not decode: {e}"),
|
|
};
|
|
Err(error_problem(StatusCode::BAD_GATEWAY, &detail))
|
|
}
|
|
|
|
/// `get_deliveries`'s query.
|
|
#[derive(Deserialize, IntoParams)]
|
|
pub struct DeliveriesQuery {
|
|
/// At most this many rows; 100 when absent, and the log keeps 500.
|
|
limit: Option<usize>,
|
|
}
|
|
|
|
/// The newest deliveries the router made, newest first.
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/notify/deliveries",
|
|
params(DeliveriesQuery),
|
|
responses(
|
|
(status = 200, description = "the router's deliveries log, newest first: when, which \
|
|
flow and node, the event and issue, who got it, and whether it went through",
|
|
body = Deliveries),
|
|
(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 log (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "notify"
|
|
)]
|
|
pub(crate) async fn get_deliveries(
|
|
State(state): State<AppState>,
|
|
Query(query): Query<DeliveriesQuery>,
|
|
) -> Result<Json<Deliveries>, ProblemDetails> {
|
|
let client = queue(&state).map_err(refused)?;
|
|
let body =
|
|
serde_json::to_vec(&DeliveriesRequest { limit: query.limit }).expect("a request encodes");
|
|
ask(&client, DELIVERIES_SUBJECT, body).await.map(Json)
|
|
}
|
|
|
|
/// The saved flows run over the newest 50 forge events, delivering nothing:
|
|
/// how many events reached each node.
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/notify/dry-run",
|
|
responses(
|
|
(status = 200, description = "per flow that would run, the number of the newest 50 \
|
|
events that reached each node", body = DryRun),
|
|
(status = 409, description = "the stored document is not one this build reads \
|
|
(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),
|
|
(status = 500, description = "the saved flows could not be read (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "notify"
|
|
)]
|
|
pub(crate) async fn get_dry_run(
|
|
State(state): State<AppState>,
|
|
) -> Result<Json<DryRun>, ProblemDetails> {
|
|
let client = queue(&state).map_err(refused)?;
|
|
let store = open(&client).await.map_err(|e| queue_problem(&e))?;
|
|
let (saved, _) = stored(&store).await.map_err(|e| queue_problem(&e))?;
|
|
let document = match saved {
|
|
Stored::Nothing => Document::default(),
|
|
Stored::Document(doc) => doc,
|
|
Stored::Unreadable(e) => return Err(unreadable(&e)),
|
|
};
|
|
let body = serde_json::to_vec(&DryRunRequest { document }).expect("a request encodes");
|
|
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};
|
|
|
|
use super::*;
|
|
use crate::test_nats::TestNats;
|
|
|
|
fn query(limit: Option<usize>) -> Query<DeliveriesQuery> {
|
|
Query(DeliveriesQuery { limit })
|
|
}
|
|
|
|
/// Answer the next request on `subject` with `reply`, handing the request
|
|
/// body back.
|
|
async fn answer(
|
|
client: &async_nats::Client,
|
|
subject: &'static str,
|
|
reply: Vec<u8>,
|
|
) -> tokio::task::JoinHandle<Vec<u8>> {
|
|
let mut requests = client.subscribe(subject).await.expect("subscribed");
|
|
client.flush().await.expect("flushed");
|
|
let client = client.clone();
|
|
tokio::spawn(async move {
|
|
let m = requests.next().await.expect("a request");
|
|
client
|
|
.publish(m.reply.expect("a reply subject"), reply.into())
|
|
.await
|
|
.expect("answered");
|
|
m.payload.to_vec()
|
|
})
|
|
}
|
|
|
|
/// A router that predates the log and the dry run has no subscriber: both
|
|
/// answer 503 with the reason, not 500 and not a hang. The control is the
|
|
/// same two calls with a router answering.
|
|
#[tokio::test]
|
|
async fn without_a_router_that_answers_both_are_unavailable() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let state = crate::agent_limits::tests::state(&client);
|
|
|
|
let deliveries = get_deliveries(State(state.clone()), query(None))
|
|
.await
|
|
.expect_err("nobody answers");
|
|
assert_eq!(deliveries.status, Some(StatusCode::SERVICE_UNAVAILABLE));
|
|
assert!(
|
|
deliveries
|
|
.detail
|
|
.as_deref()
|
|
.unwrap_or_default()
|
|
.contains("older build"),
|
|
"{deliveries:?}"
|
|
);
|
|
let dry = get_dry_run(State(state.clone()))
|
|
.await
|
|
.expect_err("nobody answers");
|
|
assert_eq!(dry.status, Some(StatusCode::SERVICE_UNAVAILABLE));
|
|
assert!(
|
|
dry.detail
|
|
.as_deref()
|
|
.unwrap_or_default()
|
|
.contains("older build"),
|
|
"{dry:?}"
|
|
);
|
|
|
|
// Control: with a router answering, the same calls are 200.
|
|
let rows = answer(&client, DELIVERIES_SUBJECT, br#"{"rows":[]}"#.to_vec()).await;
|
|
let Json(got) = get_deliveries(State(state.clone()), query(None))
|
|
.await
|
|
.expect("answered");
|
|
assert!(got.rows.is_empty());
|
|
rows.await.expect("the router task");
|
|
let hits = DryRun {
|
|
events: 0,
|
|
flows: Vec::new(),
|
|
};
|
|
let dry_run = answer(&client, DRY_RUN_SUBJECT, serde_json::to_vec(&hits).unwrap()).await;
|
|
let Json(got) = get_dry_run(State(state)).await.expect("answered");
|
|
assert_eq!(got, hits);
|
|
dry_run.await.expect("the router task");
|
|
}
|
|
|
|
/// No queue on this host is 503 too.
|
|
#[tokio::test]
|
|
async fn without_a_queue_both_are_unavailable() {
|
|
let (state, _) = crate::routes::tests::state_with_roster();
|
|
let a = get_deliveries(State(state.clone()), query(None))
|
|
.await
|
|
.expect_err("no queue");
|
|
let b = get_dry_run(State(state)).await.expect_err("no queue");
|
|
assert_eq!(a.status, Some(StatusCode::SERVICE_UNAVAILABLE));
|
|
assert_eq!(b.status, Some(StatusCode::SERVICE_UNAVAILABLE));
|
|
}
|
|
|
|
/// The limit reaches the router, and the dry run sends the saved document.
|
|
#[tokio::test]
|
|
async fn the_limit_and_the_saved_document_reach_the_router() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let state = crate::agent_limits::tests::state(&client);
|
|
|
|
let asked = answer(&client, DELIVERIES_SUBJECT, br#"{"rows":[]}"#.to_vec()).await;
|
|
let _ = get_deliveries(State(state.clone()), query(Some(7)))
|
|
.await
|
|
.expect("answered");
|
|
let request: DeliveriesRequest =
|
|
serde_json::from_slice(&asked.await.expect("task")).expect("a request");
|
|
assert_eq!(request.limit, Some(7));
|
|
|
|
// Nothing saved: the default document.
|
|
let reply = DryRun {
|
|
events: 3,
|
|
flows: vec![FlowHits {
|
|
flow: "assigned".to_owned(),
|
|
nodes: vec![NodeHits {
|
|
node: "source".to_owned(),
|
|
hits: 3,
|
|
}],
|
|
}],
|
|
};
|
|
let asked = answer(
|
|
&client,
|
|
DRY_RUN_SUBJECT,
|
|
serde_json::to_vec(&reply).unwrap(),
|
|
)
|
|
.await;
|
|
let Json(got) = get_dry_run(State(state.clone())).await.expect("answered");
|
|
assert_eq!(got, reply);
|
|
let sent: DryRunRequest =
|
|
serde_json::from_slice(&asked.await.expect("task")).expect("a request");
|
|
assert_eq!(sent.document, Document::default());
|
|
|
|
// A saved document: the router gets that one.
|
|
let store = open(&client).await.expect("bucket");
|
|
let saved = Document {
|
|
disabled_builtins: vec!["closed".to_owned()],
|
|
..Document::default()
|
|
};
|
|
store
|
|
.put(
|
|
notify_flows::KEY,
|
|
serde_json::to_vec(&saved).unwrap().into(),
|
|
)
|
|
.await
|
|
.expect("seed");
|
|
let asked = answer(
|
|
&client,
|
|
DRY_RUN_SUBJECT,
|
|
serde_json::to_vec(&reply).unwrap(),
|
|
)
|
|
.await;
|
|
let _ = get_dry_run(State(state)).await.expect("answered");
|
|
let sent: DryRunRequest =
|
|
serde_json::from_slice(&asked.await.expect("task")).expect("a request");
|
|
assert_eq!(sent.document, saved);
|
|
}
|
|
|
|
/// A refusal from the router is 502 with its reason; an unreadable saved
|
|
/// document is 409 and never reaches the router.
|
|
#[tokio::test]
|
|
async fn a_refusal_is_502_and_an_unreadable_document_is_409() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let state = crate::agent_limits::tests::state(&client);
|
|
|
|
let router = answer(
|
|
&client,
|
|
DRY_RUN_SUBJECT,
|
|
br#"{"error":"no events"}"#.to_vec(),
|
|
)
|
|
.await;
|
|
let refused = get_dry_run(State(state.clone()))
|
|
.await
|
|
.expect_err("refused");
|
|
assert_eq!(refused.status, Some(StatusCode::BAD_GATEWAY));
|
|
assert!(
|
|
refused
|
|
.detail
|
|
.as_deref()
|
|
.unwrap_or_default()
|
|
.contains("no events")
|
|
);
|
|
router.await.expect("the router task");
|
|
|
|
let store = open(&client).await.expect("bucket");
|
|
store
|
|
.put(
|
|
notify_flows::KEY,
|
|
br#"{"version":2,"flows":[]}"#.to_vec().into(),
|
|
)
|
|
.await
|
|
.expect("seed");
|
|
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));
|
|
}
|
|
}
|