288 lines
10 KiB
Rust
288 lines
10 KiB
Rust
//! The router's two answers to swarm-controller: the newest deliveries and a
|
|
//! dry run (`swarm_wire::router_api`).
|
|
//!
|
|
//! Each is a NATS request. A dry run replays the newest forge events from the
|
|
//! events stream through [`dry_run::run`] and delivers nothing, writes no
|
|
//! snapshot and sends no review request. The router hears these on their own
|
|
//! task, so a slow replay never holds up the events.
|
|
|
|
use anyhow::Result;
|
|
use async_nats::jetstream;
|
|
use futures_util::StreamExt as _;
|
|
use serde::Serialize;
|
|
use swarm_wire::forge_events;
|
|
use swarm_wire::router_api::{
|
|
DEFAULT_ROWS, DELIVERIES_SUBJECT, DRY_RUN_EVENTS, DRY_RUN_SUBJECT, Deliveries,
|
|
DeliveriesRequest, DryRunRequest, Refused,
|
|
};
|
|
|
|
use crate::{dry_run, log};
|
|
|
|
/// Answer requests on both subjects until the connection closes.
|
|
pub async fn serve(client: async_nats::Client) {
|
|
let js = jetstream::new(client.clone());
|
|
let (mut deliveries, mut dry) = match (
|
|
client.subscribe(DELIVERIES_SUBJECT).await,
|
|
client.subscribe(DRY_RUN_SUBJECT).await,
|
|
) {
|
|
(Ok(a), Ok(b)) => (a, b),
|
|
(a, b) => {
|
|
let why = a
|
|
.err()
|
|
.map(|e| e.to_string())
|
|
.or(b.err().map(|e| e.to_string()));
|
|
tracing::error!(error = ?why, "notify router: cannot hear api requests; the controller will report them unavailable");
|
|
return;
|
|
}
|
|
};
|
|
tracing::info!("notify router: answering deliveries and dry-run requests");
|
|
loop {
|
|
let (request, dry_run) = tokio::select! {
|
|
Some(m) = deliveries.next() => (m, false),
|
|
Some(m) = dry.next() => (m, true),
|
|
else => return,
|
|
};
|
|
let (client, js) = (client.clone(), js.clone());
|
|
tokio::spawn(async move {
|
|
let Some(reply) = request.reply else { return };
|
|
let answer = if dry_run {
|
|
dry_run_answer(&js, &request.payload).await
|
|
} else {
|
|
deliveries_answer(&js, &request.payload).await
|
|
};
|
|
if let Err(e) = client.publish(reply, answer.into()).await {
|
|
tracing::warn!(error = %e, "notify router: could not answer an api request");
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
/// The reply body: `value`, or a [`Refused`] saying why not.
|
|
fn body<T: Serialize>(answer: Result<T>) -> Vec<u8> {
|
|
match answer {
|
|
Ok(value) => serde_json::to_vec(&value).expect("an answer encodes"),
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
error = format!("{e:#}"),
|
|
"notify router: an api request failed"
|
|
);
|
|
serde_json::to_vec(&Refused {
|
|
error: format!("{e:#}"),
|
|
})
|
|
.expect("a refusal encodes")
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn deliveries_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
|
|
body(
|
|
async {
|
|
let request: DeliveriesRequest = if payload.is_empty() {
|
|
DeliveriesRequest::default()
|
|
} else {
|
|
serde_json::from_slice(payload)?
|
|
};
|
|
let limit = request.limit.unwrap_or(DEFAULT_ROWS);
|
|
Ok(Deliveries {
|
|
rows: log::recent(js, limit).await?,
|
|
})
|
|
}
|
|
.await,
|
|
)
|
|
}
|
|
|
|
async fn dry_run_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
|
|
body(
|
|
async {
|
|
let request: DryRunRequest = serde_json::from_slice(payload)?;
|
|
request
|
|
.document
|
|
.check()
|
|
.map_err(|why| anyhow::anyhow!("the document fails its check: {why}"))?;
|
|
let (flows, _) = request.document.active();
|
|
let mut events: Vec<(String, Vec<u8>)> =
|
|
log::newest(js, forge_events::STREAM, DRY_RUN_EVENTS)
|
|
.await?
|
|
.into_iter()
|
|
.filter_map(|(subject, body)| {
|
|
Some((forge_events::event_type(&subject)?.to_owned(), body))
|
|
})
|
|
.collect();
|
|
events.reverse();
|
|
Ok(dry_run::run(&flows, &events))
|
|
}
|
|
.await,
|
|
)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::time::Duration;
|
|
|
|
use swarm_wire::notify_flows::Document;
|
|
use swarm_wire::router_api::{DryRun, Kind, LOG_STREAM, Row};
|
|
use swarm_wire::router_todo;
|
|
|
|
use super::*;
|
|
use crate::test_nats::TestNats;
|
|
|
|
/// Send a request to the router's api, retrying while its subscription
|
|
/// is not yet in place.
|
|
async fn ask(client: &async_nats::Client, subject: &str, body: Vec<u8>) -> Vec<u8> {
|
|
for _ in 0..50 {
|
|
if let Ok(m) = client
|
|
.request(subject.to_owned(), body.clone().into())
|
|
.await
|
|
{
|
|
return m.payload.to_vec();
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
}
|
|
panic!("nothing answered {subject}");
|
|
}
|
|
|
|
fn row(n: u64) -> Row {
|
|
log::Cause {
|
|
event: "issue_assign".to_owned(),
|
|
repo: "o/r".to_owned(),
|
|
number: n,
|
|
}
|
|
.row(
|
|
"assigned",
|
|
"sink",
|
|
Kind::Todo,
|
|
"atlas",
|
|
Ok("upsert k".to_owned()),
|
|
)
|
|
}
|
|
|
|
async fn publish_event(js: &jetstream::Context, event: &str, n: i64) {
|
|
let body = serde_json::json!({
|
|
"action": "assigned",
|
|
"repository": { "full_name": "o/r", "id": 1 },
|
|
"issue": { "number": n, "title": "t", "html_url": "u", "assignees": [{ "login": "atlas" }] },
|
|
"sender": { "login": "mara" },
|
|
});
|
|
js.publish(
|
|
forge_events::subject(event),
|
|
serde_json::to_vec(&body).unwrap().into(),
|
|
)
|
|
.await
|
|
.expect("published")
|
|
.await
|
|
.expect("acked");
|
|
}
|
|
|
|
/// The newest rows come back newest first, no more than asked for, and
|
|
/// the stream keeps 500 of them.
|
|
#[tokio::test]
|
|
async fn the_deliveries_are_the_newest_rows_and_the_log_keeps_500() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let js = jetstream::new(client.clone());
|
|
log::open_or_create(&js).await.expect("the log stream");
|
|
tokio::spawn(serve(client.clone()));
|
|
for n in 1..=3 {
|
|
log::append(&js, &row(n)).await;
|
|
}
|
|
|
|
let all: Deliveries =
|
|
serde_json::from_slice(&ask(&client, DELIVERIES_SUBJECT, Vec::new()).await)
|
|
.expect("an answer");
|
|
let numbers: Vec<u64> = all.rows.iter().map(|r| r.number).collect();
|
|
assert_eq!(numbers, [3, 2, 1]);
|
|
let first = Row {
|
|
at: all.rows[0].at,
|
|
..row(3)
|
|
};
|
|
assert_eq!(all.rows[0], first);
|
|
|
|
let two: Deliveries = serde_json::from_slice(
|
|
&ask(&client, DELIVERIES_SUBJECT, br#"{"limit":2}"#.to_vec()).await,
|
|
)
|
|
.expect("an answer");
|
|
assert_eq!(
|
|
two.rows.iter().map(|r| r.number).collect::<Vec<_>>(),
|
|
[3, 2]
|
|
);
|
|
|
|
for n in 4..=503 {
|
|
log::append(&js, &row(n)).await;
|
|
}
|
|
let mut stream = js.get_stream(LOG_STREAM).await.expect("the stream");
|
|
assert_eq!(stream.info().await.expect("info").state.messages, 500);
|
|
let newest: Deliveries = serde_json::from_slice(
|
|
&ask(&client, DELIVERIES_SUBJECT, br#"{"limit":9999}"#.to_vec()).await,
|
|
)
|
|
.expect("an answer");
|
|
assert_eq!(newest.rows.len(), 500);
|
|
assert_eq!(newest.rows[0].number, 503);
|
|
assert_eq!(newest.rows[499].number, 4);
|
|
}
|
|
|
|
/// A dry run counts the newest 50 events, and delivers nothing: no todo
|
|
/// op, no log row.
|
|
#[tokio::test]
|
|
async fn a_dry_run_counts_the_newest_fifty_events_and_delivers_nothing() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let js = jetstream::new(client.clone());
|
|
swarm_queue_client::forge_events::open_or_create(&client)
|
|
.await
|
|
.expect("events");
|
|
swarm_queue_client::router_todo::open_or_create(&client)
|
|
.await
|
|
.expect("todos");
|
|
log::open_or_create(&js).await.expect("the log stream");
|
|
tokio::spawn(serve(client.clone()));
|
|
// 60 assign events on distinct issues, then one on a type no flow
|
|
// starts from.
|
|
for n in 1..=60 {
|
|
publish_event(&js, "issue_assign", n).await;
|
|
}
|
|
publish_event(&js, "issue_comment", 61).await;
|
|
|
|
let request = serde_json::to_vec(&DryRunRequest {
|
|
document: Document::default(),
|
|
})
|
|
.unwrap();
|
|
let reply = ask(&client, DRY_RUN_SUBJECT, request).await;
|
|
let done: DryRun = serde_json::from_slice(&reply).expect("a dry run");
|
|
assert_eq!(done.events, 50);
|
|
let hits = |flow: &str, node: &str| {
|
|
done.flows
|
|
.iter()
|
|
.find(|f| f.flow == flow)
|
|
.and_then(|f| f.nodes.iter().find(|n| n.node == node))
|
|
.map(|n| n.hits)
|
|
};
|
|
// 49 of the newest 50 are assign events; the 50th is the comment.
|
|
assert_eq!(hits("assigned", "source"), Some(49));
|
|
assert_eq!(hits("assigned", "sink"), Some(49));
|
|
assert_eq!(hits("unassigned", "sink"), Some(0));
|
|
|
|
for (stream, what) in [(router_todo::STREAM, "todo op"), (LOG_STREAM, "log row")] {
|
|
let mut s = js.get_stream(stream).await.expect("the stream");
|
|
assert_eq!(s.info().await.expect("info").state.messages, 0, "a {what}");
|
|
}
|
|
}
|
|
|
|
/// A document that fails its check is refused, and says why.
|
|
#[tokio::test]
|
|
async fn a_document_that_fails_its_check_is_refused() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
tokio::spawn(serve(client.clone()));
|
|
let doc = Document {
|
|
disabled_builtins: vec!["no-such-flow".to_owned()],
|
|
..Document::default()
|
|
};
|
|
let request = serde_json::to_vec(&DryRunRequest { document: doc }).unwrap();
|
|
let refused: Refused =
|
|
serde_json::from_slice(&ask(&client, DRY_RUN_SUBJECT, request).await)
|
|
.expect("a refusal");
|
|
assert!(refused.error.contains("no-such-flow"), "{}", refused.error);
|
|
let garbled = ask(&client, DRY_RUN_SUBJECT, b"nonsense".to_vec()).await;
|
|
assert!(serde_json::from_slice::<Refused>(&garbled).is_ok());
|
|
}
|
|
}
|